""" 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 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 block if present import re think_match = re.search(r'(.*?)', 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' }