File size: 7,137 Bytes
a80a31b
57cb63a
a80a31b
 
 
57cb63a
 
a80a31b
 
 
 
57cb63a
a80a31b
 
57cb63a
 
680e484
57cb63a
32d65bf
a80a31b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
57cb63a
a80a31b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
57cb63a
 
 
 
 
 
 
 
 
 
 
 
 
a80a31b
57cb63a
 
a80a31b
57cb63a
 
 
a80a31b
 
 
 
 
57cb63a
 
 
 
 
 
 
 
 
 
 
 
 
 
a80a31b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
57cb63a
 
a80a31b
57cb63a
 
a80a31b
57cb63a
a80a31b
57cb63a
 
a80a31b
57cb63a
a80a31b
 
 
 
57cb63a
 
a80a31b
 
 
 
 
 
 
 
 
 
 
 
 
b96d38b
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
import base64
import requests
import asyncio
from typing import Dict, Any
import logging
import json
import os

logger = logging.getLogger(__name__)

class LabReportAnalyzer:
    """Lab Report Analysis service using Google AI Studio API"""
    
    def __init__(self):
        """Initialize the analyzer with Google AI Studio API"""
        # Try to get API key from environment variable first, fallback to hardcoded
        self.api_key = os.getenv("GOOGLE_AI_STUDIO_API_KEY", "AIzaSyBiPSbIhtIFHwdmpJU9dwf79sqfu07BcHY")
        self.base_url = "https://generativelanguage.googleapis.com/v1beta/models"
        self.model = "gemini-2.5-flash-lite"  # Using Gemini 2.0 Flash for vision capabilities
        
    async def analyze_report(self, image_b64: str) -> Dict[str, Any]:
        """
        Analyze a lab report image and return structured results
        
        Args:
            image_b64: Base64 encoded image string
            
        Returns:
            Dictionary containing structured analysis results
        """
        try:
            prompt = self._get_analysis_prompt()
            
            # Run the inference in a thread pool to avoid blocking
            loop = asyncio.get_event_loop()
            completion = await loop.run_in_executor(
                None, 
                self._run_inference, 
                image_b64, 
                prompt
            )
            
            # Extract and parse the response
            analysis_text = completion.get('candidates', [{}])[0].get('content', {}).get('parts', [{}])[0].get('text', '').strip()
            
            # Parse the structured response
            parsed_result = self._parse_analysis_result(analysis_text)
            
            return parsed_result
            
        except Exception as e:
            logger.error(f"Error in analyze_report: {str(e)}")
            return {
                "error": True,
                "message": f"Analysis failed: {str(e)}",
                "raw_response": ""
            }
    
    def _run_inference(self, image_b64: str, prompt: str):
        """Run the Google AI Studio API inference synchronously"""
        url = f"{self.base_url}/{self.model}:generateContent"
        
        headers = {
            "Content-Type": "application/json",
        }
        
        params = {
            "key": self.api_key
        }
        
        payload = {
            "contents": [
                {
                    "parts": [
                        {"text": prompt},
                        {
                            "inline_data": {
                                "mime_type": "image/jpeg",
                                "data": image_b64
                            }
                        }
                    ]
                }
            ],
            "generationConfig": {
                "temperature": 0.1,
                "topP": 0.8,
                "topK": 10,
                "maxOutputTokens": 2048,
            }
        }
        
        response = requests.post(url, headers=headers, params=params, json=payload, timeout=30)
        
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"API request failed with status {response.status_code}: {response.text}")
    
    def _get_analysis_prompt(self) -> str:
        """Get the structured analysis prompt"""
        return """
You are a medical analysis assistant.

Analyze the following lab report image and give a structured, professional summary
following these steps:

1. Extract the results (with normal ranges if available).
2. Highlight abnormal values clearly.
3. Explain what the results suggest in simple terms.
4. Provide an overall summary of health findings.
5. End with the disclaimer:
   "This analysis is for educational purposes only and should not replace professional medical advice."

If the image is unreadable, respond: "The image text is unclear."

Format your response as follows:

Summary: (2–3 sentences explaining what the report shows)
Key Findings: (3–5 bullet points with main abnormal or notable values)
Interpretation: (1–2 sentences explaining what the findings suggest)
Note: (One line disclaimer that it's not medical advice)

Keep it short, clear, and professional — like a medical summary written for quick review.
"""
    
    def _parse_analysis_result(self, analysis_text: str) -> Dict[str, Any]:
        """
        Parse the structured analysis result into a dictionary
        
        Args:
            analysis_text: Raw analysis text from the model
            
        Returns:
            Structured dictionary with parsed components
        """
        try:
            result = {
                "error": False,
                "summary": "",
                "key_findings": [],
                "interpretation": "",
                "note": "",
                "raw_response": analysis_text
            }
            
            # Check if image is unreadable
            if "The image text is unclear" in analysis_text:
                result["error"] = True
                result["message"] = "The image text is unclear or unreadable"
                return result
            
            lines = analysis_text.split('\n')
            current_section = None
            
            for line in lines:
                line = line.strip()
                if not line:
                    continue
                    
                # Identify sections
                if line.startswith('Summary:'):
                    current_section = 'summary'
                    result['summary'] = line.replace('Summary:', '').strip()
                elif line.startswith('Key Findings:'):
                    current_section = 'key_findings'
                elif line.startswith('Interpretation:'):
                    current_section = 'interpretation'
                    result['interpretation'] = line.replace('Interpretation:', '').strip()
                elif line.startswith('Note:'):
                    current_section = 'note'
                    result['note'] = line.replace('Note:', '').strip()
                else:
                    # Continue previous section
                    if current_section == 'summary' and not result['summary']:
                        result['summary'] = line
                    elif current_section == 'key_findings' and line.startswith(('•', '-', '*')):
                        result['key_findings'].append(line.lstrip('•-* '))
                    elif current_section == 'interpretation' and not result['interpretation']:
                        result['interpretation'] = line
                    elif current_section == 'note' and not result['note']:
                        result['note'] = line
            
            return result
            
        except Exception as e:
            logger.error(f"Error parsing analysis result: {str(e)}")
            return {
                "error": True,
                "message": f"Failed to parse analysis: {str(e)}",
                "raw_response": analysis_text
            }