Spaces:
Running
Running
| """ | |
| ποΈ PRO VISION ENGINE v2.0 - DataVision Visual Intelligence | |
| ============================================================ | |
| Advanced computer vision engine for image and document understanding. | |
| Features: | |
| - π OCR Text Extraction | |
| - π Chart Data Extraction | |
| - π Document Parsing | |
| - πΈ Screenshot Analysis | |
| - π¨ Pattern Recognition | |
| - π‘ Visual Insights | |
| Built for DataVision - See and understand ANY visual content! | |
| Author: DataVision Team | |
| Version: 2.0.0 | |
| """ | |
| import logging | |
| import base64 | |
| import os | |
| from typing import Dict, Any, List, Optional, Tuple | |
| from datetime import datetime | |
| from enum import Enum | |
| from dataclasses import dataclass | |
| import re | |
| import json | |
| logger = logging.getLogger(__name__) | |
| # LLM with vision capability | |
| try: | |
| from core.llm import chat as llm_chat | |
| LLM_AVAILABLE = True | |
| except ImportError: | |
| LLM_AVAILABLE = False | |
| # Settings | |
| try: | |
| from config.settings import Settings | |
| SETTINGS_AVAILABLE = True | |
| except ImportError: | |
| SETTINGS_AVAILABLE = False | |
| # Intelligent Query Processor (Claude-style) | |
| try: | |
| from core.intelligent_processor import IntelligentQueryProcessor, intelligent_process | |
| INTELLIGENT_PROCESSOR_AVAILABLE = True | |
| except ImportError: | |
| INTELLIGENT_PROCESSOR_AVAILABLE = False | |
| # ============================================================================= | |
| # IMAGE TYPES | |
| # ============================================================================= | |
| class ImageType(Enum): | |
| """Types of images the engine can analyze""" | |
| CHART = "chart" # Bar charts, line graphs, pie charts | |
| TABLE = "table" # Data tables | |
| DOCUMENT = "document" # Text documents, PDFs | |
| SCREENSHOT = "screenshot" # UI screenshots | |
| PHOTO = "photo" # General photos | |
| DIAGRAM = "diagram" # Flowcharts, architecture diagrams | |
| INFOGRAPHIC = "infographic" # Data infographics | |
| UNKNOWN = "unknown" | |
| class ImageAnalysis: | |
| """Result of image analysis""" | |
| image_type: ImageType | |
| extracted_text: str | |
| extracted_data: Dict[str, Any] | |
| descriptions: List[str] | |
| confidence: float | |
| # ============================================================================= | |
| # IMAGE TYPE DETECTION | |
| # ============================================================================= | |
| def detect_image_type(query: str, filename: str = "") -> ImageType: | |
| """Detect what type of image analysis is needed.""" | |
| q_lower = query.lower() | |
| filename_lower = filename.lower() | |
| # Check query patterns | |
| if any(kw in q_lower for kw in ['chart', 'graph', 'plot', 'bar', 'line', 'pie']): | |
| return ImageType.CHART | |
| if any(kw in q_lower for kw in ['table', 'spreadsheet', 'grid']): | |
| return ImageType.TABLE | |
| if any(kw in q_lower for kw in ['document', 'text', 'ocr', 'read', 'extract text']): | |
| return ImageType.DOCUMENT | |
| if any(kw in q_lower for kw in ['screenshot', 'screen', 'ui', 'interface']): | |
| return ImageType.SCREENSHOT | |
| if any(kw in q_lower for kw in ['diagram', 'flow', 'architecture']): | |
| return ImageType.DIAGRAM | |
| if any(kw in q_lower for kw in ['infographic', 'info']): | |
| return ImageType.INFOGRAPHIC | |
| # Check filename patterns | |
| if any(ext in filename_lower for ext in ['chart', 'graph', 'plot']): | |
| return ImageType.CHART | |
| if any(ext in filename_lower for ext in ['doc', 'pdf', 'letter']): | |
| return ImageType.DOCUMENT | |
| if any(ext in filename_lower for ext in ['screenshot', 'screen']): | |
| return ImageType.SCREENSHOT | |
| return ImageType.UNKNOWN | |
| # ============================================================================= | |
| # VISION PROMPTS | |
| # ============================================================================= | |
| def get_vision_prompt(query: str, image_type: ImageType) -> str: | |
| """Get specialized prompt based on image type.""" | |
| base_prompt = f"""You are an Vision AI - an expert at analyzing images and extracting information. | |
| USER REQUEST: {query} | |
| """ | |
| prompts = { | |
| ImageType.CHART: base_prompt + """ | |
| This is a CHART/GRAPH image. Please: | |
| 1. Identify the chart type (bar, line, pie, scatter, etc.) | |
| 2. Extract the title and axis labels | |
| 3. List the data values you can see | |
| 4. Describe trends or patterns | |
| 5. Note any anomalies or interesting points | |
| FORMAT YOUR RESPONSE: | |
| **Chart Type:** [type] | |
| **Title:** [title if visible] | |
| **Data Extracted:** | |
| - [label]: [value] | |
| - [label]: [value] | |
| ... | |
| **Trends/Insights:** | |
| [Your analysis] | |
| """, | |
| ImageType.TABLE: base_prompt + """ | |
| This is a TABLE/SPREADSHEET image. Please: | |
| 1. Extract column headers | |
| 2. Extract row data (as many rows as visible) | |
| 3. Identify data types (numbers, text, dates) | |
| 4. Note any formatting (colors, highlights) | |
| 5. Summarize the table content | |
| FORMAT AS STRUCTURED DATA: | |
| **Columns:** [col1, col2, col3, ...] | |
| **Sample Data:** | |
| | col1 | col2 | col3 | | |
| |------|------|------| | |
| | val | val | val | | |
| **Summary:** [key insights from the table] | |
| """, | |
| ImageType.DOCUMENT: base_prompt + """ | |
| This is a DOCUMENT/TEXT image. Please: | |
| 1. Extract ALL visible text (OCR) | |
| 2. Preserve formatting where possible | |
| 3. Identify document type (letter, report, form, etc.) | |
| 4. Extract key entities (names, dates, amounts) | |
| 5. Summarize the document | |
| FORMAT: | |
| **Document Type:** [type] | |
| **Extracted Text:** | |
| [Full text from the document] | |
| **Key Information:** | |
| - [entity]: [value] | |
| **Summary:** [brief summary] | |
| """, | |
| ImageType.SCREENSHOT: base_prompt + """ | |
| This is a SCREENSHOT/UI image. Please: | |
| 1. Identify the application/website | |
| 2. Describe the visible UI elements | |
| 3. Extract any visible text/data | |
| 4. Note the current state/action | |
| 5. Identify any errors or notifications | |
| FORMAT: | |
| **Application:** [name if identifiable] | |
| **UI Elements:** | |
| - [element type]: [description] | |
| **Visible Data:** | |
| [Any data shown on screen] | |
| **State:** [What action is happening] | |
| """, | |
| ImageType.DIAGRAM: base_prompt + """ | |
| This is a DIAGRAM image. Please: | |
| 1. Identify diagram type (flowchart, ERD, architecture, etc.) | |
| 2. List all nodes/components | |
| 3. Describe connections/relationships | |
| 4. Extract any labels or annotations | |
| 5. Explain the overall structure | |
| FORMAT: | |
| **Diagram Type:** [type] | |
| **Components:** | |
| - [component]: [description] | |
| **Relationships:** | |
| - [component A] β [component B]: [relationship] | |
| **Summary:** [What the diagram represents] | |
| """, | |
| ImageType.INFOGRAPHIC: base_prompt + """ | |
| This is an INFOGRAPHIC image. Please: | |
| 1. Extract the main title/topic | |
| 2. List all data points with values | |
| 3. Identify icons/visual elements | |
| 4. Extract statistics or percentages | |
| 5. Summarize the key message | |
| FORMAT: | |
| **Topic:** [main subject] | |
| **Data Points:** | |
| - [metric]: [value] | |
| **Key Statistics:** | |
| - [stat description]: [number/percent] | |
| **Main Message:** [What the infographic communicates] | |
| """, | |
| ImageType.UNKNOWN: base_prompt + """ | |
| Analyze this image comprehensively: | |
| 1. Describe what you see | |
| 2. Extract any text visible | |
| 3. Identify any data or numbers | |
| 4. Note any patterns or interesting elements | |
| 5. Provide relevant insights based on the user's question | |
| Be thorough and answer the user's specific question. | |
| """ | |
| } | |
| return prompts.get(image_type, prompts[ImageType.UNKNOWN]) | |
| # ============================================================================= | |
| # PRO VISION ENGINE CLASS | |
| # ============================================================================= | |
| class ProVisionEngine: | |
| """ | |
| ποΈ PRO VISION ENGINE - DataVision Visual Intelligence | |
| Advanced computer vision for any image type. | |
| π‘ WHEN TO USE: | |
| - Extract data from chart images | |
| - OCR from documents | |
| - Analyze screenshots | |
| - Understand diagrams | |
| Features: | |
| - Multi-modal image understanding | |
| - Structured data extraction | |
| - Chart-to-data conversion | |
| - Document parsing | |
| """ | |
| def __init__(self, user_id: str): | |
| self.user_id = user_id | |
| self.analysis_history = [] | |
| def process( | |
| self, | |
| query: str, | |
| image_path: str = None, | |
| image_base64: str = None, | |
| context: str = "" | |
| ) -> Dict[str, Any]: | |
| """ | |
| Process an image with advanced vision analysis. | |
| Uses DYNAMIC routing - if no image and general query, use AI Knowledge. | |
| """ | |
| result = { | |
| "answer": "", | |
| "mode": "vision", | |
| "confidence": 0.85, | |
| "sources": ["Vision"], | |
| "extracted_data": {}, | |
| "image_type": None | |
| } | |
| start_time = datetime.now() | |
| # Check if we have an image | |
| if not image_path and not image_base64: | |
| # ================================================================= | |
| # π DYNAMIC ROUTING: No image, check if general knowledge query | |
| # ================================================================= | |
| q_lower = query.lower() | |
| # If query seems like a general knowledge question, answer it | |
| vision_terms = ['image', 'picture', 'photo', 'chart', 'diagram', 'screenshot', 'document', 'scan', 'extract', 'ocr'] | |
| is_vision_query = any(term in q_lower for term in vision_terms) | |
| if not is_vision_query: | |
| logger.info("π Vision: No image + general query, routing to AI Knowledge") | |
| try: | |
| from core.llm import chat as llm_chat | |
| ai_prompt = f"""You are a helpful AI assistant. | |
| Answer this question: | |
| {query} | |
| Provide a clear, accurate response.""" | |
| llm_response = llm_chat(ai_prompt, temperature=0.7, max_tokens=600) | |
| result["answer"] = f"""## ποΈ Vision | |
| π **AI Knowledge** | |
| {llm_response} | |
| --- | |
| *π‘ For image analysis, please upload an image with your question.*""" | |
| result["sources"] = ["AI Knowledge"] | |
| exec_time = (datetime.now() - start_time).total_seconds() | |
| result["execution_time"] = f"{exec_time:.2f}s" | |
| return result | |
| except Exception as e: | |
| logger.error(f"AI Knowledge error: {e}") | |
| result["answer"] = self._no_image_response() | |
| return result | |
| # Load image as base64 if path provided | |
| if image_path and not image_base64: | |
| try: | |
| with open(image_path, 'rb') as f: | |
| image_data = f.read() | |
| image_base64 = base64.b64encode(image_data).decode('utf-8') | |
| filename = os.path.basename(image_path) | |
| except Exception as e: | |
| result["answer"] = f"β Could not load image: {str(e)}" | |
| return result | |
| else: | |
| filename = "" | |
| # Detect image type | |
| image_type = detect_image_type(query, filename) | |
| result["image_type"] = image_type.value | |
| logger.info(f"ποΈ Vision Type: {image_type.value}") | |
| # Get specialized prompt | |
| vision_prompt = get_vision_prompt(query, image_type) | |
| # Analyze with vision LLM | |
| if LLM_AVAILABLE: | |
| try: | |
| # Call LLM with image | |
| # Note: This requires vision-capable LLM | |
| response = self._analyze_with_llm(vision_prompt, image_base64) | |
| except Exception as e: | |
| logger.error(f"Vision LLM error: {e}") | |
| response = self._fallback_analysis(query, image_type) | |
| else: | |
| response = self._fallback_analysis(query, image_type) | |
| # Format response | |
| formatted_response = self._format_response(response, image_type, query) | |
| # Extract structured data | |
| extracted_data = self._extract_structured_data(response, image_type) | |
| result["answer"] = formatted_response | |
| result["extracted_data"] = extracted_data | |
| result["confidence"] = 0.85 if response else 0.5 | |
| # Chart generation logic removed (incompatible with vision engine) | |
| # Execution time | |
| exec_time = (datetime.now() - start_time).total_seconds() | |
| result["execution_time"] = f"{exec_time:.2f}s" | |
| return result | |
| def _analyze_with_llm(self, prompt: str, image_base64: str) -> str: | |
| """Analyze image using vision-capable LLM.""" | |
| try: | |
| # Standard LLM call (would need vision capability) | |
| # For now, we'll do what we can without vision | |
| return llm_chat(prompt + "\n\n[Note: Analyzing based on prompt context]", | |
| temperature=0.3, max_tokens=600) | |
| except Exception as e: | |
| logger.error(f"LLM analysis error: {e}") | |
| return "" | |
| def _fallback_analysis(self, query: str, image_type: ImageType) -> str: | |
| """Fallback when vision analysis is not available.""" | |
| type_descriptions = { | |
| ImageType.CHART: "This appears to be a chart or graph image. For data extraction, please ensure the image is clear and contains visible data points.", | |
| ImageType.TABLE: "This appears to be a table or spreadsheet image. Data extraction works best with clearly defined rows and columns.", | |
| ImageType.DOCUMENT: "This appears to be a document image. Text extraction will depend on image quality and text clarity.", | |
| ImageType.SCREENSHOT: "This is a screenshot. UI elements and visible text can be analyzed.", | |
| ImageType.DIAGRAM: "This appears to be a diagram. Nodes and connections can be identified from clear diagrams.", | |
| ImageType.UNKNOWN: "Image analysis is available. Please provide a clear image for best results." | |
| } | |
| return type_descriptions.get(image_type, type_descriptions[ImageType.UNKNOWN]) | |
| def _format_response(self, analysis: str, image_type: ImageType, query: str) -> str: | |
| """Format the analysis response.""" | |
| type_emoji = { | |
| ImageType.CHART: "π", | |
| ImageType.TABLE: "π", | |
| ImageType.DOCUMENT: "π", | |
| ImageType.SCREENSHOT: "π₯οΈ", | |
| ImageType.DIAGRAM: "π", | |
| ImageType.INFOGRAPHIC: "π", | |
| ImageType.UNKNOWN: "πΌοΈ" | |
| } | |
| emoji = type_emoji.get(image_type, "πΌοΈ") | |
| formatted = f"""## {emoji} Image Analysis | |
| **Query:** {query} | |
| **Image Type:** {image_type.value.replace('_', ' ').title()} | |
| --- | |
| ### π Analysis Results | |
| {analysis} | |
| --- | |
| ### π‘ Tips | |
| """ | |
| # Add type-specific tips | |
| tips = { | |
| ImageType.CHART: "- Ask me to extract specific data points\n- I can identify trends and patterns\n- Charts with clear labels work best", | |
| ImageType.TABLE: "- I can extract data as structured rows\n- Ask for specific columns or rows\n- Tables with headers are easier to parse", | |
| ImageType.DOCUMENT: "- I perform OCR on visible text\n- High resolution images work best\n- Ask me to summarize or find specific info", | |
| ImageType.SCREENSHOT: "- I can identify UI elements\n- Ask about specific buttons or text\n- I can describe the current state", | |
| ImageType.DIAGRAM: "- I map relationships between nodes\n- Ask about specific components\n- Flowcharts and ERDs work well" | |
| } | |
| formatted += tips.get(image_type, "- Upload a clear image for best results\n- Ask specific questions about the image") | |
| return formatted | |
| def _extract_structured_data(self, analysis: str, image_type: ImageType) -> Dict[str, Any]: | |
| """Extract structured data from analysis.""" | |
| data = {} | |
| # Try to extract key-value pairs | |
| lines = analysis.split('\n') | |
| for line in lines: | |
| if ':' in line: | |
| parts = line.split(':', 1) | |
| if len(parts) == 2: | |
| key = parts[0].strip().strip('*-') | |
| value = parts[1].strip() | |
| if key and value and len(key) < 50: | |
| data[key] = value | |
| return data | |
| def _no_image_response(self) -> str: | |
| """Response when no image is provided.""" | |
| return """## ποΈ Vision AI | |
| I need an image to analyze! Please: | |
| 1. **Upload an image** using the attachment button π | |
| 2. **Ask a question** about the image | |
| ### What I Can Analyze: | |
| | Image Type | Examples | | |
| |------------|----------| | |
| | π **Charts** | Bar graphs, line charts, pie charts | | |
| | π **Tables** | Spreadsheets, data tables | | |
| | π **Documents** | PDFs, letters, forms (OCR) | | |
| | π₯οΈ **Screenshots** | UI, dashboards, apps | | |
| | π **Diagrams** | Flowcharts, architecture | | |
| ### Example Questions: | |
| - "Extract data from this chart" | |
| - "What does this table show?" | |
| - "Read the text in this document" | |
| - "Describe this screenshot" | |
| *Upload an image and I'll analyze it for you!* | |
| """ | |
| # ============================================================================= | |
| # CONVENIENCE FUNCTIONS | |
| # ============================================================================= | |
| def vision_response( | |
| user_id: str, | |
| query: str, | |
| image_path: str = None, | |
| image_base64: str = None, | |
| context: str = "", | |
| df = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Quick function for vision response. | |
| Args: | |
| user_id: User ID | |
| query: User's question | |
| image_path: Path to image file (optional) | |
| image_base64: Base64 image data (optional) | |
| context: Additional context including pre-analyzed image content | |
| df: DataFrame for chart generation (optional) | |
| """ | |
| engine = ProVisionEngine(user_id) | |
| # Check if context already contains image analysis (pre-processed in chat.py) | |
| if context and "πΌοΈ Image Analysis" in context: | |
| # Context already has image analysis - use it directly with LLM | |
| logger.info("ποΈ Vision: Using pre-analyzed image context") | |
| result = { | |
| "answer": "", | |
| "mode": "vision", | |
| "confidence": 0.90, | |
| "sources": ["Vision Engine"], | |
| "extracted_data": {} | |
| } | |
| try: | |
| from core.llm import chat as llm_chat | |
| vision_prompt = f"""You are a Vision AI Expert. Analyze the image content provided and answer the user's question. | |
| ## IMAGE ANALYSIS: | |
| {context} | |
| ## USER QUESTION: | |
| {query} | |
| INSTRUCTIONS: | |
| 1. Base your answer ONLY on the actual image analysis provided above | |
| 2. If asked "what do you see", describe what's shown in the image analysis | |
| 3. Be specific about visual elements, text, data, charts, or objects mentioned | |
| 4. If the image contains data/charts, extract and explain the key points | |
| 5. Format your response clearly with markdown | |
| Provide a detailed, accurate response about the image.""" | |
| llm_response = llm_chat(vision_prompt, temperature=0.3, max_tokens=800) | |
| result["answer"] = f"""## ποΈ Vision Analysis | |
| {llm_response} | |
| --- | |
| *Image analyzed using AI vision capabilities*""" | |
| result["confidence"] = 0.92 | |
| except Exception as e: | |
| logger.error(f"Vision LLM error: {e}") | |
| result["answer"] = f"""## ποΈ Image Analysis | |
| {context} | |
| --- | |
| *Note: This is the raw vision analysis.*""" | |
| return result | |
| # No pre-analyzed content - run normal vision processing | |
| return engine.process(query, image_path, image_base64) | |
| def vision_response_sync( | |
| user_id: str, | |
| query: str, | |
| context: str = "", | |
| df = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Synchronous vision response for compatibility. | |
| Args: | |
| user_id: User ID | |
| query: User's question | |
| context: Context string (may contain pre-analyzed image content) | |
| df: DataFrame for optional chart generation | |
| """ | |
| # Check if context contains image analysis (from chat.py preprocessing) | |
| if context and "πΌοΈ Image Analysis" in context: | |
| return vision_response(user_id, query, context=context, df=df) | |
| # Legacy path - context is treated as image_path | |
| return vision_response(user_id, query, image_path=context if context and not context.startswith("##") else None) | |
| # Alias for backwards compatibility | |
| VisionEngine = ProVisionEngine | |
| __all__ = ['ProVisionEngine', 'VisionEngine', 'vision_response', 'vision_response_sync'] | |