Spaces:
Sleeping
Sleeping
File size: 12,449 Bytes
5ccd893 4739805 5ccd893 207961e 5ccd893 207961e 5ccd893 4739805 207961e 4739805 207961e 4739805 207961e 4739805 207961e 4739805 207961e 4739805 5ccd893 | 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 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 | """
Chat Controller
Handles chat-related business logic and service coordination
"""
import logging
from typing import Dict, Any, Optional
from flask import request
from services.ai_service import AIService
from services.gee_service import GEEService
class ChatController:
"""Controller for chat operations"""
def __init__(self, ai_service: AIService, gee_service: GEEService):
self.ai_service = ai_service
self.gee_service = gee_service
self.logger = logging.getLogger(__name__)
def handle_chat_message_streaming(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Process chat message — same logic as handle_chat_message but returns
a flat dict suitable for the SSE streaming generator.
"""
return self.handle_chat_message(data)
def handle_chat_message(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Process incoming chat message and generate response
Args:
data: Request data containing message and optional context
Returns:
Response dictionary
"""
try:
# Extract message from request data
message = data.get('message', '').strip()
if not message:
return {
'error': 'Message is required',
'status': 'error'
}
# Extract optional context
context = data.get('context', {})
# Check if satellite data is requested
location = context.get('location')
if location and isinstance(location, dict):
lat = location.get('latitude')
lon = location.get('longitude')
if lat is not None and lon is not None:
# Add satellite data to context
context = self._enrich_context_with_satellite_data(context, lat, lon)
conversation_history = data.get('history') or []
# Generate AI response
response = self.ai_service.generate_response(
message, context, conversation_history=conversation_history
)
if response['status'] == 'success':
raw_text = response['message'].strip()
# Clean potential markdown wrappers and <think> blocks
json_str = raw_text
start_idx = raw_text.find('{')
end_idx = raw_text.rfind('}')
if start_idx != -1 and end_idx != -1 and end_idx > start_idx:
json_str = raw_text[start_idx:end_idx+1]
try:
import json
parsed = json.loads(json_str)
gee_code = parsed.get('gee_code', None)
if gee_code is not None and str(gee_code).strip().lower() in ('null', 'none', ''):
gee_code = None
vis_params = parsed.get('vis_params', None)
if vis_params is not None and str(vis_params).strip().lower() in ('null', 'none', ''):
vis_params = None
response_text = parsed.get('response_text', '')
thinking = parsed.get('thinking', '')
# Extract external <think> block if present
import re
think_match = re.search(r'<think>(.*?)</think>', raw_text, re.DOTALL)
if think_match:
extracted_think = think_match.group(1).strip()
if thinking:
thinking = extracted_think + "\n\n" + thinking
else:
thinking = extracted_think
center = parsed.get('center', None)
tile_url = None
gee_error = None
# Try to execute GEE code if provided and GEE is available
if gee_code and vis_params:
if self.gee_service.initialized:
gee_result = self.gee_service.execute_gee_code_with_fallback(
gee_code, vis_params, user_message=message
)
if gee_result.get('status') == 'success':
tile_url = gee_result.get('tile_url')
if gee_result.get('used_fallback'):
response_text = (
response_text
+ " (Rendered with a standard index recipe after the "
"initial code was invalid.)"
)
else:
gee_error = gee_result.get('message', 'GEE execution failed')
self.logger.warning(f"GEE execution error: {gee_error}")
elif self.gee_service.initialized:
# Model returned no code — try template from user message (e.g. "NBI of Kerala")
fallback_only = self.gee_service.execute_gee_code_with_fallback(
'', {}, user_message=message
)
if fallback_only.get('status') == 'success':
tile_url = fallback_only.get('tile_url')
gee_error = None
else:
gee_error = 'GEE service not initialized'
self.logger.warning("GEE code received but GEE service not initialized")
return {
'status': 'success',
'message': response_text,
'thinking': thinking,
'tile_url': tile_url,
'gee_code': gee_code,
'vis_params': vis_params,
'legend': parsed.get('legend', None),
'center': center,
'gee_error': gee_error,
'metadata': {
'model': response.get('model'),
'attempt': response.get('attempt'),
'context_enriched': bool(context)
}
}
except Exception as parse_err:
self.logger.error(f"Failed to parse AI JSON response: {parse_err}. Raw text: {raw_text[:200]}")
return {
'response': raw_text,
'message': raw_text,
'status': 'success',
'metadata': {
'model': response.get('model'),
'attempt': response.get('attempt'),
'context_enriched': bool(context)
}
}
else:
return {
'error': response.get('message', 'Failed to generate response'),
'status': 'error'
}
except Exception as e:
self.logger.error(f"Chat message processing error: {str(e)}")
return {
'error': f'Internal server error: {str(e)}',
'status': 'error'
}
def _enrich_context_with_satellite_data(self, context: Dict[str, Any], lat: float, lon: float) -> Dict[str, Any]:
"""
Enrich context with satellite data if GEE is available
Args:
context: Current context
lat: Latitude
lon: Longitude
Returns:
Enriched context
"""
try:
if self.gee_service.initialized:
# Get recent satellite data (last 30 days)
from datetime import datetime, timedelta
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
satellite_data = self.gee_service.get_satellite_data(
latitude=lat,
longitude=lon,
start_date=start_date.strftime('%Y-%m-%d'),
end_date=end_date.strftime('%Y-%m-%d')
)
context['satellite_data'] = satellite_data
except Exception as e:
self.logger.warning(f"Failed to enrich context with satellite data: {str(e)}")
return context
def analyze_location(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Analyze a specific location for disaster indicators
Args:
data: Request data containing location and analysis parameters
Returns:
Analysis results
"""
try:
# Extract location data
latitude = data.get('latitude')
longitude = data.get('longitude')
if latitude is None or longitude is None:
return {
'error': 'Latitude and longitude are required',
'status': 'error'
}
# Get satellite data
from datetime import datetime, timedelta
end_date = datetime.now()
start_date = end_date - timedelta(days=data.get('days_back', 30))
satellite_data = self.gee_service.get_satellite_data(
latitude=latitude,
longitude=longitude,
start_date=start_date.strftime('%Y-%m-%d'),
end_date=end_date.strftime('%Y-%m-%d'),
cloud_filter=data.get('cloud_filter', 20)
)
# Get AI analysis
user_query = data.get('query', 'Analyze this location for potential disaster indicators')
analysis = self.ai_service.analyze_satellite_data(satellite_data, user_query)
return {
'status': 'success',
'location': {
'latitude': latitude,
'longitude': longitude
},
'satellite_data': satellite_data,
'analysis': analysis,
'parameters': {
'days_back': data.get('days_back', 30),
'cloud_filter': data.get('cloud_filter', 20)
}
}
except Exception as e:
self.logger.error(f"Location analysis error: {str(e)}")
return {
'error': f'Analysis failed: {str(e)}',
'status': 'error'
}
def get_disaster_info(self, disaster_type: str, location_data: Dict[str, Any] = None) -> Dict[str, Any]:
"""
Get information about a specific disaster type
Args:
disaster_type: Type of disaster
location_data: Optional location context
Returns:
Disaster information
"""
try:
# Validate disaster type
valid_disasters = ['flood', 'drought', 'storm', 'landslide', 'wildfire', 'earthquake']
if disaster_type.lower() not in valid_disasters:
return {
'error': f'Invalid disaster type. Valid types: {", ".join(valid_disasters)}',
'status': 'error'
}
# Get AI insights
insights = self.ai_service.get_disaster_insights(disaster_type, location_data)
return {
'status': 'success',
'disaster_type': disaster_type,
'insights': insights,
'location_specific': bool(location_data)
}
except Exception as e:
self.logger.error(f"Disaster info error: {str(e)}")
return {
'error': f'Failed to get disaster information: {str(e)}',
'status': 'error'
} |