multimodal_tools.py

"""
Multimodal Analysis Tools using OpenRouter Vision Models
"""
import os
import base64
import tempfile
import requests
import numpy as np
from langchain_core.tools import tool


def call_openrouter_vision(
    model: str,
    question: str,
    image_base64: str = None,
    fallback_model: str = None
) -> str:
    """
    Call OpenRouter vision model for image analysis.
    
    Args:
        model: Model ID (e.g.,  "qwen/qwen3-vl-30b-a3b-thinking")
        question: Question about the image  
        image_base64: Base64 encoded image
        fallback_model: Fallback model if primary fails
        
    Returns:
        Model's response text
    """
    api_key = os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        raise ValueError("OPENROUTER_API_KEY not found in environment")
    
    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": question}
            ]
        }
    ]
    
    # Add image if provided
    if image_base64:
        messages[0]["content"].append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{image_base64}"
            }
        })
    
    try:
        response = requests.post(
            "https://openrouter.ai/api/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": model,
                "messages": messages,
                "max_tokens": 2048
            },
            timeout=60
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]
    except Exception as e:
        if fallback_model and fallback_model != model:
            print(f"Primary model {model} failed, trying fallback {fallback_model}: {e}")
            return call_openrouter_vision(fallback_model, question, image_base64)
        raise Exception(f"OpenRouter vision call failed: {e}")


@tool
def vision_analyze_image(question: str, image_path: str) -> str:
    """
    Analyze an image using AI vision model to answer questions about it.
    Use this for semantic understanding of images (chess positions, charts, diagrams, screenshots, etc.)
    
    Args:
        question: Question about the image
        image_path: Path to image file
        
    Returns:
        Analysis result from vision model
    """
    try:
        # Load and encode image
        with open(image_path, "rb") as f:
            image_data = base64.b64encode(f.read()).decode("utf-8")
        
        # Call OpenRouter vision model with fallback
        result = call_openrouter_vision(
            model="qwen/qwen3-vl-30b-a3b-thinking",
            question=question,
            image_base64=image_data,
            fallback_model="google/gemini-2.5-flash"
        )
        return result
    except Exception as e:
        return f"Error analyzing image: {str(e)}"


import subprocess
import glob

# ... (call_openrouter_vision remains same)

@tool
def vision_analyze_video(question: str, video_path: str, num_frames: int = 5) -> str:
    """
    Analyze a video file by extracting key frames using FFmpeg.
    
    Args:
        question: Question about the video
        video_path: Path to video file
        num_frames: Number of frames to extract
        
    Returns:
        Analysis result combining insights from all frames
    """
    try:
        with tempfile.TemporaryDirectory() as tmpdir:
            # Use FFmpeg to extract frames at intervals
            # fps=1/interval? Easier: just extract 5 frames uniformly?
            # Strategy: Extract 5 frames at percentage intervals (0%, 20%, 40%...)
            
            # First, extract frames
            subprocess.run([
                "ffmpeg", "-i", video_path, 
                "-vf", f"fps={num_frames}/(duration)", # approximate 
                # Better: select='not(mod(n,1000))' is hard without duration.
                # Simplest: vf fps=1 to get 1 per second, then take N
                "-vf", "fps=1", 
                os.path.join(tmpdir, "frame_%03d.jpg")
            ], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            
            # List generated frames
            frames = sorted(glob.glob(os.path.join(tmpdir, "*.jpg")))
            
            if not frames:
                # Fallback: try extracting just 5 frames total using 'select' filter
                # or just 1 frame if short
                subprocess.run([
                   "ffmpeg", "-i", video_path, "-vframes", "5", 
                   os.path.join(tmpdir, "thumb%d.jpg")
                ], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                frames = sorted(glob.glob(os.path.join(tmpdir, "*.jpg")))

            # Pick num_frames evenly spaced
            if len(frames) > num_frames:
                indices = np.linspace(0, len(frames)-1, num_frames, dtype=int)
                selected_frames = [frames[i] for i in indices]
            else:
                selected_frames = frames

            frames_analysis = []
            for idx, frame_path in enumerate(selected_frames):
                with open(frame_path, "rb") as f:
                    frame_b64 = base64.b64encode(f.read()).decode("utf-8")
                
                frame_question = f"Frame {idx+1}: {question}"
                analysis = call_openrouter_vision(
                    model="qwen/qwen3-vl-30b-a3b-thinking",
                    question=frame_question,
                    image_base64=frame_b64,
                    fallback_model="google/gemini-2.5-flash"
                )
                frames_analysis.append(f"Frame {idx+1}: {analysis}")
            
            combined = "\n\n".join(frames_analysis)
            return f"Video analysis ({len(selected_frames)} frames extracted via FFmpeg):\n{combined}"

    except Exception as e:
        return f"Error analyzing video: {str(e)}"




@tool  
def vision_analyze_document(question: str, file_path: str) -> str:
    """
    Analyze a document (TXT/MD) using AI.
    For PDF or other formats, please use Code Interpreter to extract text or convert to images first.
    
    Args:
        question: Question about the document
        file_path: Path to document file
        
    Returns:
        Analysis result from document content
    """
    try:
        text_content = ""
        
        # Extract text based on file type
        if file_path.lower().endswith(('.txt', '.md', '.py', '.js', '.json', '.html', '.css', '.csv')):
            with open(file_path, 'r', encoding='utf-8') as f:
                text_content = f.read()
            
            # Analyze with Gemini (good for documents)
            result = call_openrouter_vision(
                model="google/gemini-3-flash-preview",
                question=f"{question}\n\nDocument content:\n{text_content[:15000]}",  # Limit length
                fallback_model="google/gemini-2.5-flash"
            )
            return result
            
        else:
            return f"Direct analysis for this file type ({os.path.basename(file_path)}) is not supported directly. Please use the Code Interpreter tool to read/convert this file first."
        
    except Exception as e:
        return f"Error analyzing document: {str(e)}"