Pemahaman Video dalam Agen

Pemahaman Video untuk Agen

Memasang OpenCV untuk Ekstraksi Bingkai

# pip install opencv-python-headless
import cv2

def get_video_metadata(video_path: str) -> dict:
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise IOError(f'Cannot open video: {video_path}')

    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    duration_s = frame_count / fps if fps > 0 else 0
    cap.release()

    return {
        'fps': round(fps, 2),
        'frame_count': frame_count,
        'width': width,
        'height': height,
        'duration_seconds': round(duration_s, 2)
    }

meta = get_video_metadata('recording.mp4')
print(meta)

Mengekstrak Bingkai Setiap N Detik

import cv2
import os

def extract_frames(
    video_path: str,
    output_dir: str,
    every_n_seconds: float = 5.0
) -> list:
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(fps * every_n_seconds)
    os.makedirs(output_dir, exist_ok=True)

    saved_frames = []
    frame_idx = 0
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        if frame_idx % frame_interval == 0:
            timestamp = round(frame_idx / fps, 2)
            fname = f'{output_dir}/frame_{frame_idx:06d}_{timestamp}s.jpg'
            cv2.imwrite(fname, frame)
            saved_frames.append({'path': fname, 'timestamp': timestamp})
        frame_idx += 1

    cap.release()
    print(f'Extracted {len(saved_frames)} frames')
    return saved_frames

Deteksi Perubahan Adegan

import cv2
import numpy as np

def extract_scene_change_frames(
    video_path: str,
    output_dir: str,
    diff_threshold: float = 30.0
) -> list:
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    os.makedirs(output_dir, exist_ok=True)
    prev_gray = None
    saved = []
    frame_idx = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        if prev_gray is not None:
            diff = np.mean(np.abs(gray.astype(float) - prev_gray.astype(float)))
            if diff > diff_threshold:
                ts = round(frame_idx / fps, 2)
                fname = f'{output_dir}/scene_{frame_idx:06d}_{ts}s.jpg'
                cv2.imwrite(fname, frame)
                saved.append({'path': fname, 'timestamp': ts, 'diff': round(diff, 2)})
        prev_gray = gray
        frame_idx += 1

    cap.release()
    return saved

Mengirim Urutan Bingkai ke API Visi

import base64
import anthropic

def describe_video_frames(frame_info_list: list, query: str) -> str:
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')
    content = []
    for fi in frame_info_list:
        with open(fi['path'], 'rb') as f:
            b64 = base64.standard_b64encode(f.read()).decode('utf-8')
        content.append({
            'type': 'text',
            'text': f'Frame at {fi["timestamp"]}s:'
        })
        content.append({
            'type': 'image',
            'source': {'type': 'base64', 'media_type': 'image/jpeg', 'data': b64}
        })
    content.append({'type': 'text', 'text': query})

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': content}]
    )
    return response.content[0].text

Instruksi Penalaran Temporal

TEMPORAL_QUERY = (
    'You are analysing a video sequence. The frames above are in chronological order '
    'with their timestamps shown. Please:\n'
    '1. Describe what changed between the first and last frame\n'
    '2. Identify any notable events and when they occurred (timestamp)\n'
    '3. Summarise the overall activity shown in the video\n'
    '4. Estimate the total duration of the main activity\n'
    'Be specific about timestamps.'
)

# More targeted temporal query:
CHANGE_DETECTION_QUERY = (
    'Compare frame at {start_ts}s and frame at {end_ts}s. '
    'List all visible differences in bullet points. '
    'Categorise each change as: object_moved, object_added, '
    'object_removed, lighting_change, or camera_move.'
)

# Usage:
query = CHANGE_DETECTION_QUERY.format(start_ts=0.0, end_ts=20.0)

if __name__ == '__main__':
    print('Temporal reasoning query:')
    print(TEMPORAL_QUERY)
    print()
    print('Change detection query:')
    print(query)

Alur Pemrosesan Ringkasan Video

def summarise_video(
    video_path: str,
    every_n_seconds: float = 10.0,
    batch_size: int = 5
) -> str:
    import anthropic
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')

    # Extract frames
    frames = extract_frames(video_path, '/tmp/video_frames', every_n_seconds)

    # Process in batches
    batch_summaries = []
    for i in range(0, len(frames), batch_size):
        batch = frames[i:i + batch_size]
        ts_range = f'{batch[0]["timestamp"]}s - {batch[-1]["timestamp"]}s'
        query = f'Summarise what happens in this video segment ({ts_range}).'
        summary = describe_video_frames(batch, query)
        batch_summaries.append(f'[{ts_range}] {summary}')

    # Synthesise
    all_summaries = '\n\n'.join(batch_summaries)
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content':
            f'Here are segment-level video summaries:\n\n{all_summaries}\n\n'
            'Write a single coherent summary of the entire video.'
        }]
    )
    return result.content[0].text

Melacak Objek Antarbingkai

def track_object_across_frames(
    frames: list,
    object_description: str
) -> list:
    import anthropic, base64
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')
    trajectory = []

    for fi in frames:
        with open(fi['path'], 'rb') as f:
            b64 = base64.standard_b64encode(f.read()).decode('utf-8')
        response = client.messages.create(
            model='claude-opus-4-5',
            max_tokens=128,
            messages=[{
                'role': 'user',
                'content': [
                    {'type': 'image', 'source': {'type': 'base64',
                      'media_type': 'image/jpeg', 'data': b64}},
                    {'type': 'text', 'text':
                      f'Where is the {object_description} in this frame? '
                      'Answer with one of: top-left, top-right, centre, '
                      'bottom-left, bottom-right, not-visible.'}
                ]
            }]
        )
        trajectory.append({
            'timestamp': fi['timestamp'],
            'position': response.content[0].text.strip()
        })
    return trajectory

Menangani Batas Jumlah Bingkai

MAX_FRAMES_PER_REQUEST = 8

def hierarchical_video_analysis(frames: list, final_query: str) -> str:
    import anthropic
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')

    # Level 1: analyse each chunk
    chunk_summaries = []
    for i in range(0, len(frames), MAX_FRAMES_PER_REQUEST):
        chunk = frames[i:i + MAX_FRAMES_PER_REQUEST]
        ts = f'{chunk[0]["timestamp"]}s-{chunk[-1]["timestamp"]}s'
        summary = describe_video_frames(
            chunk, f'What happens in this segment ({ts})?'
        )
        chunk_summaries.append(f'Segment {ts}: {summary}')

    # Level 2: synthesise all chunk summaries
    combined = '\n'.join(chunk_summaries)
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{
            'role': 'user',
            'content': f'Video segments:\n{combined}\n\n{final_query}'
        }]
    )
    return result.content[0].text

Menggabungkan Analisis Audio dan Video

def full_video_analysis(video_path: str) -> dict:
    import subprocess

    # Step 1: Extract audio track
    audio_path = '/tmp/video_audio.mp3'
    subprocess.run([
        'ffmpeg', '-i', video_path, '-q:a', '0', '-map', 'a',
        audio_path, '-y'
    ], capture_output=True)

    # Step 2: Transcribe audio
    transcript_data = transcribe_with_timestamps(audio_path)

    # Step 3: Extract key frames
    frames = extract_frames(video_path, '/tmp/key_frames', every_n_seconds=15.0)

    # Step 4: Visual summary
    visual_summary = hierarchical_video_analysis(
        frames, 'Summarise the visual content.'
    )

    return {
        'transcript': transcript_data['full_text'],
        'transcript_segments': transcript_data['segments'],
        'visual_summary': visual_summary,
        'frame_count': len(frames)
    }

Pembersihan dan Pengelolaan Bingkai

import os
import shutil
import hashlib

FRAME_CACHE_DIR = '/tmp/frame_cache'

def get_cache_key(video_path: str, every_n_seconds: float) -> str:
    content = f'{video_path}:{every_n_seconds}'
    return hashlib.md5(content.encode()).hexdigest()[:12]

def get_or_extract_frames(video_path: str, every_n_seconds: float) -> list:
    key = get_cache_key(video_path, every_n_seconds)
    cache_dir = os.path.join(FRAME_CACHE_DIR, key)

    if os.path.exists(cache_dir):
        print(f'Cache hit: {key}')
        files = sorted(os.listdir(cache_dir))
        return [
            {'path': os.path.join(cache_dir, f),
             'timestamp': float(f.split('_')[-1].replace('s.jpg', ''))}
            for f in files if f.endswith('.jpg')
        ]
    return extract_frames(video_path, cache_dir, every_n_seconds)

def clear_frame_cache():
    if os.path.exists(FRAME_CACHE_DIR):
        shutil.rmtree(FRAME_CACHE_DIR)
        print('Frame cache cleared')

Uji Pemahaman

Ringkasan: Pemahaman Video dalam Agen