Spaces:
Sleeping
Sleeping
| """ | |
| 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' | |
| } |