query_vision.py

"""

Vision-based query module using GPT-5 Vision.

Supports multimodal queries combining text and images.

"""

import base64
import json
import logging
import sqlite3
from typing import List, Tuple, Optional, Dict, Any
import numpy as np
from PIL import Image
from openai import OpenAI

from config import *
from utils import ImageProcessor, classify_image

logger = logging.getLogger(__name__)

class VisionRetriever:
    """Vision-based retrieval using GPT-5 Vision for image analysis and classification."""
    
    def __init__(self):
        self.client = OpenAI(api_key=OPENAI_API_KEY)
        self.image_processor = ImageProcessor()
    
    def get_similar_images(self, query_image_path: str, top_k: int = 5) -> List[Dict[str, Any]]:
        """Find similar images in the database based on classification similarity."""
        try:
            # Uses GPT-5 Vision for classification-based similarity search
            # Note: This implementation uses classification similarity rather than embeddings
            
            # Classify the query image
            query_classification = classify_image(query_image_path)
            
            # Query database for similar images
            conn = sqlite3.connect(IMAGES_DB)
            cursor = conn.cursor()
            
            # Search for images with similar classification
            cursor.execute("""

                SELECT image_id, image_path, classification, metadata

                FROM images 

                WHERE classification LIKE ?

                ORDER BY created_at DESC

                LIMIT ?

            """, (f"%{query_classification}%", top_k))
            
            results = cursor.fetchall()
            conn.close()
            
            similar_images = []
            for row in results:
                image_id, image_path, classification, metadata_json = row
                metadata = json.loads(metadata_json) if metadata_json else {}
                
                similar_images.append({
                    'image_id': image_id,
                    'image_path': image_path,
                    'classification': classification,
                    'metadata': metadata,
                    'similarity_score': 0.8  # Classification-based similarity score
                })
            
            logger.info(f"Found {len(similar_images)} similar images for query")
            return similar_images
            
        except Exception as e:
            logger.error(f"Error finding similar images: {e}")
            return []
    
    def analyze_image_safety(self, image_path: str, question: str = None) -> str:
        """Analyze image for safety concerns using GPT-5 Vision."""
        try:
            # Convert image to base64
            with open(image_path, "rb") as image_file:
                image_b64 = base64.b64encode(image_file.read()).decode()
            
            # Create analysis prompt
            if question:
                analysis_prompt = (
                    f"Analyze this image in the context of the following question: {question}\n\n"
                    "Please provide a detailed safety analysis covering:\n"
                    "1. What equipment, machinery, or workplace elements are visible\n"
                    "2. Any potential safety hazards or compliance issues\n"
                    "3. Relevant OSHA standards or regulations that may apply\n"
                    "4. Recommendations for safety improvements\n"
                    "5. How this relates to the specific question asked"
                )
            else:
                analysis_prompt = (
                    "Analyze this image for occupational safety and health concerns. Provide:\n"
                    "1. Description of what's shown in the image\n"
                    "2. Identification of potential safety hazards\n"
                    "3. Relevant OSHA standards or safety regulations\n"
                    "4. Recommendations for improving safety"
                )
            
            messages = [{
                "role": "user",
                "content": [
                    {"type": "text", "text": analysis_prompt},
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}", "detail": "high"}}
                ]
            }]
            
            # For GPT-5 vision, temperature must be default (1.0) and reasoning is not supported
            response = self.client.chat.completions.create(
                model=OPENAI_CHAT_MODEL,
                messages=messages,
                max_completion_tokens=DEFAULT_MAX_TOKENS
            )
            
            return response.choices[0].message.content.strip()
            
        except Exception as e:
            logger.error(f"Error analyzing image: {e}")
            return f"I encountered an error while analyzing the image: {e}"
    
    def retrieve_relevant_text(self, image_classification: str, question: str, top_k: int = 3) -> List[Dict[str, Any]]:
        """Retrieve text documents relevant to the image classification and question."""
        # This would integrate with other retrieval methods to find relevant text
        # For now, we'll create a simple keyword-based search
        
        try:
            # Import other query modules for text retrieval
            from query_vanilla import query as vanilla_query
            
            # Create an enhanced query combining image classification and original question
            enhanced_question = f"safety requirements for {image_classification} {question}"
            
            # Use vanilla retrieval to find relevant text
            _, text_citations = vanilla_query(enhanced_question, top_k=top_k)
            
            return text_citations
            
        except Exception as e:
            logger.error(f"Error retrieving relevant text: {e}")
            return []
    
    def generate_multimodal_answer(self, question: str, image_analysis: str, 

                                 text_citations: List[Dict], similar_images: List[Dict]) -> str:
        """Generate answer combining image analysis and text retrieval."""
        try:
            # Prepare context from text citations
            text_context = ""
            if text_citations:
                text_parts = []
                for i, citation in enumerate(text_citations, 1):
                    if 'text' in citation:
                        text_parts.append(f"[Text Source {i}] {citation['source']}: {citation['text'][:500]}...")
                    else:
                        text_parts.append(f"[Text Source {i}] {citation['source']}")
                text_context = "\n\n".join(text_parts)
            
            # Prepare context from similar images
            image_context = ""
            if similar_images:
                image_parts = []
                for img in similar_images[:3]:  # Limit to top 3
                    source = img['metadata'].get('source', 'Unknown')
                    classification = img.get('classification', 'unknown')
                    image_parts.append(f"Similar image from {source}: classified as {classification}")
                image_context = "\n".join(image_parts)
            
            # Create comprehensive prompt
            system_message = (
                "You are an expert in occupational safety and health. "
                "You have been provided with an image analysis, relevant text documents, "
                "and information about similar images in the database. "
                "Provide a comprehensive answer that integrates all this information."
            )
            
            user_message = f"""Question: {question}



Image Analysis:

{image_analysis}



Relevant Text Documentation:

{text_context}



Similar Images Context:

{image_context}



Please provide a comprehensive answer that:

1. Addresses the specific question asked

2. Incorporates insights from the image analysis

3. References relevant regulatory information from the text sources

4. Notes any connections to similar cases or images

5. Provides actionable recommendations based on safety standards"""
            
            # For GPT-5, temperature must be default (1.0) and reasoning is not supported
            response = self.client.chat.completions.create(
                model=OPENAI_CHAT_MODEL,
                messages=[
                    {"role": "system", "content": system_message},
                    {"role": "user", "content": user_message}
                ],
                max_completion_tokens=DEFAULT_MAX_TOKENS * 2  # Allow longer response for comprehensive analysis
            )
            
            return response.choices[0].message.content.strip()
            
        except Exception as e:
            logger.error(f"Error generating multimodal answer: {e}")
            return "I apologize, but I encountered an error while generating the comprehensive answer."

# Global retriever instance
_retriever = None

def get_retriever() -> VisionRetriever:
    """Get or create global vision retriever instance."""
    global _retriever
    if _retriever is None:
        _retriever = VisionRetriever()
    return _retriever

def query(question: str, image_path: Optional[str] = None, top_k: int = DEFAULT_TOP_K) -> Tuple[str, List[Dict]]:
    """

    Main vision-based query function with unified signature.

    

    Args:

        question: User question

        image_path: Path to image file (required for vision queries)

        top_k: Number of relevant results to retrieve

        

    Returns:

        Tuple of (answer, citations)

    """
    if not image_path:
        return "Vision queries require an image. Please provide an image file.", []
    
    try:
        retriever = get_retriever()
        
        # Step 1: Analyze the provided image
        logger.info(f"Analyzing image: {image_path}")
        image_analysis = retriever.analyze_image_safety(image_path, question)
        
        # Step 2: Classify the image
        image_classification = classify_image(image_path)
        
        # Step 3: Find similar images
        similar_images = retriever.get_similar_images(image_path, top_k=3)
        
        # Step 4: Retrieve relevant text documents
        text_citations = retriever.retrieve_relevant_text(image_classification, question, top_k)
        
        # Step 5: Generate comprehensive multimodal answer
        answer = retriever.generate_multimodal_answer(
            question, image_analysis, text_citations, similar_images
        )
        
        # Step 6: Prepare citations
        citations = []
        
        # Add image analysis as primary citation
        citations.append({
            'rank': 1,
            'type': 'image_analysis',
            'source': f"Analysis of {image_path.split('/')[-1] if '/' in image_path else image_path.split('\\')[-1]}",
            'method': 'vision',
            'classification': image_classification,
            'score': 1.0
        })
        
        # Add text citations
        for i, citation in enumerate(text_citations, 2):
            citation_copy = citation.copy()
            citation_copy['rank'] = i
            citation_copy['method'] = 'vision_text'
            citations.append(citation_copy)
        
        # Add similar images
        for i, img in enumerate(similar_images):
            citations.append({
                'rank': len(citations) + 1,
                'type': 'similar_image',
                'source': img['metadata'].get('source', 'Image Database'),
                'method': 'vision',
                'classification': img.get('classification', 'unknown'),
                'similarity_score': img.get('similarity_score', 0.0),
                'image_id': img.get('image_id')
            })
        
        logger.info(f"Vision query completed. Generated {len(citations)} citations.")
        return answer, citations
        
    except Exception as e:
        logger.error(f"Error in vision query: {e}")
        error_message = "I apologize, but I encountered an error while processing your vision-based question."
        return error_message, []

def query_image_only(image_path: str, question: str = None) -> Tuple[str, List[Dict]]:
    """

    Analyze image without text retrieval (faster for simple image analysis).

    

    Args:

        image_path: Path to image file

        question: Optional specific question about the image

        

    Returns:

        Tuple of (analysis, citations)

    """
    try:
        retriever = get_retriever()
        
        # Analyze image
        analysis = retriever.analyze_image_safety(image_path, question)
        
        # Classify image
        classification = classify_image(image_path)
        
        # Create citation for image analysis
        citations = [{
            'rank': 1,
            'type': 'image_analysis',
            'source': f"Analysis of {image_path.split('/')[-1] if '/' in image_path else image_path.split('\\')[-1]}",
            'method': 'vision_only',
            'classification': classification,
            'score': 1.0
        }]
        
        return analysis, citations
        
    except Exception as e:
        logger.error(f"Error in image-only analysis: {e}")
        return "Error analyzing image.", []

def query_with_details(question: str, image_path: Optional[str] = None, 

                      top_k: int = DEFAULT_TOP_K) -> Tuple[str, List[Dict], List[Tuple]]:
    """

    Vision query function that returns detailed chunk information (for compatibility).

    

    Returns:

        Tuple of (answer, citations, chunks)

    """
    answer, citations = query(question, image_path, top_k)
    
    # Convert citations to chunk format for backward compatibility
    chunks = []
    for citation in citations:
        if citation['type'] == 'image_analysis':
            chunks.append((
                f"Image Analysis ({citation['classification']})",
                citation['score'],
                "Analysis of uploaded image for safety compliance",
                citation['source']
            ))
        elif citation['type'] == 'similar_image':
            chunks.append((
                f"Similar Image (Score: {citation.get('similarity_score', 0):.3f})",
                citation.get('similarity_score', 0),
                f"Similar image classified as {citation['classification']}",
                citation['source']
            ))
        else:
            chunks.append((
                f"Text Reference {citation['rank']}",
                citation.get('score', 0.5),
                citation.get('text', 'Referenced document'),
                citation['source']
            ))
    
    return answer, citations, chunks

if __name__ == "__main__":
    # Test the vision system (requires an actual image file)
    import sys
    
    if len(sys.argv) > 1:
        test_image_path = sys.argv[1]
        test_question = "What safety issues can you identify in this image?"
        
        print("Testing vision retrieval system...")
        print(f"Image: {test_image_path}")
        print(f"Question: {test_question}")
        print("-" * 50)
        
        try:
            answer, citations = query(test_question, test_image_path)
            
            print("Answer:")
            print(answer)
            print(f"\nCitations ({len(citations)}):")
            for citation in citations:
                print(f"- {citation['source']} (Type: {citation.get('type', 'unknown')})")
                
        except Exception as e:
            print(f"Error during testing: {e}")
    else:
        print("To test vision system, provide an image path as argument:")
        print("python query_vision.py /path/to/image.jpg")