File size: 3,511 Bytes
6062397
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
"""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)