core/image_analysis.py

"""Image analysis using Claude Vision.

Provides infrastructure image analysis for the FixMyNeighborhood app.
"""

import base64
from typing import Optional
import anthropic

from config import ANTHROPIC_API_KEY


# Initialize Claude client for image analysis
_claude_client: Optional[anthropic.Anthropic] = None


def get_claude_client() -> Optional[anthropic.Anthropic]:
    """Get or create the Claude client for image analysis."""
    global _claude_client
    if _claude_client is None and ANTHROPIC_API_KEY:
        try:
            _claude_client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
            print("Claude client initialized for image analysis")
        except Exception as e:
            print(f"Claude client error: {e}")
    return _claude_client


class ImageAnalyzer:
    """
    Analyzes infrastructure images using Claude Vision.

    Provides concise analysis of:
    - Issue type (pothole, streetlight, drain, etc.)
    - Severity assessment
    - Safety hazard evaluation
    """

    ANALYSIS_PROMPT = (
        "Describe this NYC infrastructure issue. What type of issue is it? "
        "How severe does it appear? Is it a safety hazard? Be concise."
    )

    def __init__(self, client: anthropic.Anthropic = None):
        self.client = client or get_claude_client()

    def analyze(self, image_path: str) -> Optional[str]:
        """
        Analyze an infrastructure image.

        Args:
            image_path: Path to the uploaded image

        Returns:
            Analysis text or None if failed
        """
        if not self.client or not image_path:
            return None

        try:
            with open(image_path, "rb") as f:
                data = base64.standard_b64encode(f.read()).decode("utf-8")

            media_type = self._get_media_type(image_path)

            response = self.client.messages.create(
                model="claude-haiku-4-5-20251001",  # Cost-optimized
                max_tokens=500,
                messages=[{
                    "role": "user",
                    "content": [
                        {
                            "type": "image",
                            "source": {
                                "type": "base64",
                                "media_type": media_type,
                                "data": data
                            }
                        },
                        {
                            "type": "text",
                            "text": self.ANALYSIS_PROMPT
                        }
                    ]
                }]
            )
            return response.content[0].text

        except Exception as e:
            print(f"Vision analysis error: {e}")
            return None

    def _get_media_type(self, image_path: str) -> str:
        """Determine media type from file extension."""
        path_lower = image_path.lower()
        if path_lower.endswith(".png"):
            return "image/png"
        elif path_lower.endswith(".gif"):
            return "image/gif"
        elif path_lower.endswith(".webp"):
            return "image/webp"
        return "image/jpeg"


# Convenience function for backwards compatibility
def analyze_image(image_path: str) -> Optional[str]:
    """
    Analyze an infrastructure image using Claude Vision.

    Args:
        image_path: Path to the uploaded image

    Returns:
        Analysis text or None if failed
    """
    analyzer = ImageAnalyzer()
    return analyzer.analyze(image_path)