import gradio as gr from datetime import datetime import sys import threading import os from agents.orchestrator import ClimateRiskOrchestrator from tools.mapping_utils import ( COUNTRIES_AND_CITIES, US_STATES, get_coordinates_from_dropdown, create_risk_map, get_city_suggestions, ) # === LogCatcher === class LogCatcher: def __init__(self): self.buffer = "" self.lock = threading.Lock() self._stdout = sys.stdout self._stderr = sys.stderr def write(self, msg): with self.lock: self.buffer += msg self._stdout.write(msg) def flush(self): pass def get_logs(self): with self.lock: return self.buffer def clear(self): with self.lock: self.buffer = "" def redirect(self): sys.stdout = self sys.stderr = self def restore(self): sys.stdout = self._stdout sys.stderr = self._stderr def isatty(self): return False def fileno(self): return self._stdout.fileno() logcatcher = LogCatcher() logcatcher.redirect() class ClimateRiskUI: """User interface for the climate risk system with dropdown and map functionality.""" def __init__(self, model): self.orchestrator = ClimateRiskOrchestrator(model) self.theme = gr.themes.Soft( primary_hue="blue", secondary_hue="gray", neutral_hue="slate" ) def update_business_visibility(self, profile_type): show_business = profile_type == "Business Owner" return gr.Dropdown(visible=show_business) def validate_and_update_api_key(self, api_key, nasa_key=""): """Validate and update API keys in environment variables.""" status_messages = [] # Validate Anthropic API key if api_key and api_key.strip(): if api_key.startswith('sk-ant-'): os.environ['ANTHROPIC_API_KEY'] = api_key.strip() status_messages.append("✅ Anthropic API key updated successfully!") # Try to reinitialize the model with new key try: from config import model # This would require reloading the model, but for now just update env status_messages.append("ℹ️ Restart the application to use the new API key.") except Exception as e: status_messages.append(f"⚠️ API key saved but model reload failed: {str(e)}") else: status_messages.append("❌ Invalid Anthropic API key format. Should start with 'sk-ant-'") else: status_messages.append("⚠️ Please enter a valid Anthropic API key") # Validate NASA FIRMS API key (optional) if nasa_key and nasa_key.strip(): os.environ['NASA_FIRMS_MAP_KEY'] = nasa_key.strip() status_messages.append("✅ NASA FIRMS API key updated successfully!") return "\n".join(status_messages) def get_current_api_status(self): """Get current API key status.""" anthropic_key = os.getenv('ANTHROPIC_API_KEY', '') nasa_key = os.getenv('NASA_FIRMS_MAP_KEY', '') status = [] if anthropic_key and anthropic_key != 'your-anthropic-api-key-here': masked_key = anthropic_key[:8] + "..." + anthropic_key[-4:] if len(anthropic_key) > 12 else "***" status.append(f"🔑 **Anthropic API Key:** {masked_key} (configured)") else: status.append("❌ **Anthropic API Key:** Not configured") if nasa_key and nasa_key != 'your-nasa-firms-api-key-here': masked_nasa = nasa_key[:8] + "..." + nasa_key[-4:] if len(nasa_key) > 12 else "***" status.append(f"🛰️ **NASA FIRMS Key:** {masked_nasa} (configured)") else: status.append("ℹ️ **NASA FIRMS Key:** Not configured (optional)") return "\n".join(status) def analyze_with_dropdown( self, country, city, state, profile_type, business_type, vulnerable_groups, ): logcatcher.clear() if not country or not city: return ( "Please select both country and city.", "", "", ) coords_result, validation_message = get_coordinates_from_dropdown(country, city, state) if coords_result is None: return validation_message, "", "" lat, lon = coords_result state_info = f", {state}" if state else "" location_full = f"{city}{state_info}, {country}" base_query = f"Perform a comprehensive climate risk assessment for {location_full}." profile_context = "" if profile_type.lower() == "business owner": business_detail = f" as a {business_type}" if business_type else "" profile_context = ( f" Focus on business continuity risks{business_detail}, including supply chain vulnerabilities, operational disruptions, infrastructure threats, customer safety, inventory protection, and revenue continuity. Consider industry-specific vulnerabilities and regulatory compliance requirements." ) elif profile_type.lower() == "electric utility": profile_context = " Emphasize electric utility risks including power outages, overloads, low reserve generation capacity, extreme weather impacts on electric utility assets, and catastrophic wildfires potential." elif profile_type.lower() == "emergency manager": profile_context = " Prioritize emergency management perspectives including evacuation planning, critical infrastructure vulnerabilities, community preparedness needs, and multi-hazard scenarios." else: profile_context = " Focus on residential safety, household preparedness, health impacts, and community-level risks." vulnerable_context = "" if vulnerable_groups: groups_text = ", ".join(vulnerable_groups) vulnerable_context = f" Pay special attention to impacts on vulnerable populations: {groups_text}." analysis_requirements = ( " Analyze earthquake, wildfire, flood, and extreme weather risks. Provide specific risk levels (0-100 scale), contributing factors, time horizons, and confidence levels. Include recent data and current conditions." ) user_query = base_query + profile_context + vulnerable_context + analysis_requirements user_profile = { "type": profile_type.lower(), "business_type": business_type if profile_type.lower() == "business owner" else None, "vulnerable_groups": vulnerable_groups or [], } print(f"[{datetime.now()}] Analyse : {user_query}") result = self.orchestrator.analyze_and_recommend(user_query, user_profile) if "error" in result: print(f"[ERROR] {result['error']}") return f"Error: {result['error']}", "", "" risk_summary = self._format_risk_analysis(result["risk_analysis"]) recommendations_text = self._format_recommendations(result["recommendations"], profile_type) enhanced_map = create_risk_map(lat, lon, city, country, result["risk_analysis"]) return risk_summary, recommendations_text, enhanced_map def update_map_from_location(self, country, city, state=None): if not country or not city: return "Please select both country and city.", "" coords_result, validation_message = get_coordinates_from_dropdown(country, city, state) if coords_result is None: return validation_message, "" lat, lon = coords_result risk_map = create_risk_map(lat, lon, city, country) return validation_message, risk_map def update_cities(self, country): suggestions = get_city_suggestions(country) show_state = country == "United States" country_centers = { "France": (48.8566, 2.3522), "United States": (39.8283, -98.5795), "United Kingdom": (51.5074, -0.1278), "Germany": (52.5200, 13.4050), "Japan": (35.6762, 139.6503), "Canada": (45.4215, -75.7040), "Australia": (-35.2809, 149.1300), "Italy": (41.9028, 12.4964), "Spain": (40.4168, -3.7038), "China": (39.9042, 116.4074), "India": (28.6139, 77.2090), "Brazil": (-15.7975, -47.8919), } lat, lon = country_centers.get(country, (48.8566, 2.3522)) basic_map = create_risk_map(lat, lon, f"Select a city in {country}", country) return suggestions, gr.Dropdown(visible=show_state), basic_map def analyze_user_input( self, user_query: str, profile_type: str, business_type: str, vulnerable_groups: list = None, ): logcatcher.clear() if not user_query.strip(): return ( "Please enter your climate risk question or location.", "", "
Map will appear here after analysis.
", ) user_profile = { "type": profile_type.lower(), "business_type": business_type if profile_type.lower() == "business owner" else None, "vulnerable_groups": vulnerable_groups or [], } print(f"[{datetime.now()}] Analyse: {user_query}") result = self.orchestrator.analyze_and_recommend(user_query, user_profile) if "error" in result: print(f"[ERROR] {result['error']}") return f"Error: {result['error']}", "", "" risk_summary = self._format_risk_analysis(result["risk_analysis"]) recommendations_text = self._format_recommendations(result["recommendations"], profile_type) location = result["risk_analysis"].get("location", {}) lat = location.get("lat", 0) lon = location.get("lon", 0) city = location.get("city", "Unknown") country = location.get("country", "Unknown") enhanced_map = create_risk_map(lat, lon, city, country, result["risk_analysis"]) return risk_summary, recommendations_text, enhanced_map def _format_risk_analysis(self, risk_analysis: dict) -> str: if not risk_analysis or "error" in risk_analysis: return "Risk analysis not available or failed." formatted = f"# 🌍 Climate Risk Analysis\n\n" location = risk_analysis.get("location", {}) if location: formatted += f"**Location:** {location.get('city', 'Unknown')}, {location.get('country', '')}\n" formatted += f"**Coordinates:** {location.get('lat', 0):.4f}°N, {location.get('lon', 0):.4f}°E\n\n" formatted += f"**Analysis Date:** {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n" overall = risk_analysis.get("overall_assessment", "No overall assessment available.") formatted += f"## 📊 Overall Assessment\n{overall}\n\n" risks = risk_analysis.get("risk_analysis", {}) if risks: formatted += "## 🎯 Individual Risk Assessment\n\n" for risk_name, risk_data in risks.items(): if isinstance(risk_data, dict): risk_level = risk_data.get("risk_level", 0) if risk_level > 80: emoji = "🔴" level_text = "VERY HIGH" elif risk_level > 60: emoji = "🟠" level_text = "HIGH" elif risk_level > 40: emoji = "🟡" level_text = "MODERATE" elif risk_level > 20: emoji = "🟢" level_text = "LOW" else: emoji = "⚪" level_text = "MINIMAL" formatted += f"### {emoji} {risk_name.title()} Risk\n" formatted += f"**Risk Level:** {level_text} ({risk_level}/100)\n" formatted += f"**Time Horizon:** {risk_data.get('time_horizon', 'Unknown')}\n" formatted += f"**Confidence:** {risk_data.get('confidence', 'Unknown')}\n\n" if risk_data.get("key_insights"): formatted += f"**Analysis:** {risk_data['key_insights']}\n\n" factors = risk_data.get("contributing_factors", []) if factors: formatted += f"**Key Factors:** {', '.join(factors)}\n\n" return formatted def _format_recommendations(self, recommendations: dict, profile_type: str) -> str: if not recommendations: return "No recommendations available." formatted = f"# 🎯 Personalized Recommendations for {profile_type} **[survivalist mode]**\n\n" if "emergency" in recommendations: formatted += "## 🚨 Emergency Preparedness\n" for rec in recommendations["emergency"]: formatted += f"- {rec}\n" formatted += "\n" if "household" in recommendations: formatted += "## 🏠 Household Adaptations\n" for rec in recommendations["household"]: formatted += f"- {rec}\n" formatted += "\n" if "business" in recommendations: formatted += "## 🏢 Business Continuity\n" for rec in recommendations["business"]: formatted += f"- {rec}\n" formatted += "\n" if "financial" in recommendations: formatted += "## 💰 Financial Planning\n" for rec in recommendations["financial"]: formatted += f"- {rec}\n" formatted += "\n" formatted += "---\n" formatted += "*Recommendations generated by AI agents based on current risk analysis and your profile.*" return formatted def create_interface(self): def get_logs(): return logcatcher.get_logs() with gr.Blocks( theme=self.theme, title="🛰️ CAVA-AI – Agentic AI based Climate Adaption & Vulnerability Assessment Tool" ) as app: gr.Markdown( """ # 🛰️ CAVA-AI – Agentic AI based Climate Adaption & Vulnerability Assessment Tool
🤖 What does CAVA-AI do?

CAVA-AI's AI agents instantly analyze climate risks ( 🌪️ Weather, 🌊 Flood, 🌍 Earthquake, 🔥 Wildfire, 🌫️ Air quality, 📈 Climate trends, ☀️ Solar radiation, 🌊 Marine forecast ) for a specified location, providing you with clear, actionable recommendations.

Analysis is fully automated, always up to date, and based on leading data sources: OpenStreetMap 🗺️, Open-Meteo 🌦️, USGS 🌎, NASA FIRMS 🔥.

How to use CAVA-AI?
Use the quick location selection (dropdowns and map) 🌍, or ask complex, personalized questions in natural language 💬.
""" ) with gr.Tabs(): with gr.TabItem("📍 Quick Location Selection"): with gr.Row(): with gr.Column(): country_dropdown = gr.Dropdown( choices=list(COUNTRIES_AND_CITIES.keys()), label="Select Country", value="United States", interactive=True, ) city_input = gr.Textbox( label="Enter City Name", placeholder="e.g., Los Angeles, San Francisco, San Diego, ...", value="Pomona", interactive=True, info="Enter any city name in the selected country", ) state_dropdown = gr.Dropdown( choices=US_STATES, label="Select State (US only)", value="California", visible=False, interactive=True, info="Select state for US locations", ) city_suggestions = gr.Markdown( get_city_suggestions("Los Angeles"), visible=True ) with gr.Column(): profile_dropdown = gr.Dropdown( choices=[ "General Public", "Business Owner", "Electric Utility", "Emergency Manager", ], label="Your Profile", value="General Public", ) vulnerable_groups = gr.CheckboxGroup( choices=[ "Elderly", "Children", "Chronic Health Conditions", "Pregnant", ], label="Vulnerable Groups in Household", ) business_type_dropdown = gr.Dropdown( choices=[ "Restaurant/Food Service", "Retail Store", "Manufacturing", "Construction", "Healthcare Facility", "Educational Institution", "Technology/Software", "Transportation/Logistics", "Tourism/Hospitality", "Financial Services", "Real Estate", "Agriculture/Farming", "Energy/Utilities", "Entertainment/Events", "Professional Services", "Small Office", "Warehouse/Distribution", "Other", ], label="Business Type", value="Retail Store", visible=False, interactive=True, info="Select your business type for specialized recommendations", ) with gr.Row(): analyze_location_btn = gr.Button( "🔍 Analyze This Location", variant="primary", size="lg" ) with gr.Row(): gr.HTML("""

🛰️ Agentic Logs

""") with gr.Row(): logs_box = gr.Textbox( value=logcatcher.get_logs(), label="Logs", lines=17, max_lines=25, interactive=False, elem_id="terminal_logs", show_copy_button=True, container=False, ) logs_timer = gr.Timer(0.5) logs_timer.tick(get_logs, None, logs_box) with gr.Row(): location_map = gr.HTML( create_risk_map(47.7486, -3.3667, "Lorient", "France"), label="Interactive Risk Map", ) with gr.Row(): location_status = gr.Markdown("", visible=True) # Résumé d'analyse dans un cadre custom (CSS) with gr.Row(): dropdown_risk_summary = gr.Markdown( "Select a location above to begin analysis.", label="Risk Assessment Summary", elem_id="risk_summary_box", ) # Recommandations dans un cadre custom (CSS) with gr.Row(): dropdown_recommendations = gr.Markdown( "Recommendations will appear here after analysis.", label="AI-Generated Recommendations", elem_id="recommendations_box", ) with gr.TabItem("💬 Natural Language Query"): with gr.Row(): with gr.Column(scale=2): user_query = gr.Textbox( label="Your Climate Risk Question", placeholder="Will Southern California experience more wildfires this summer?", lines=3, info="Be as specific as possible about location, timeframe, and what you're concerned about.", ) gr.Markdown( """ **Examples:** - "What are the wildfire risks in Southern California this summer?" - "I live in San Joaquin Valley, can I expect power outages this week ?" - "I'm planning to move to Long Beach, what climate risks should I be aware of?" - "How should I prepare for climate change?" - "What ermergency preparations should SCE make for possible earthquakes?" """ ) with gr.Column(scale=1): nl_profile_type = gr.Dropdown( choices=[ "General Public", "Business Owner", "Electric Utility", "Emergency Manager", ], label="Your Profile", value="General Public", ) nl_business_type_dropdown = gr.Dropdown( choices=[ "Restaurant/Food Service", "Retail Store", "Manufacturing", "Construction", "Healthcare Facility", "Educational Institution", "Technology/Software", "Transportation/Logistics", "Tourism/Hospitality", "Financial Services", "Real Estate", "Agriculture/Farming", "Energy/Utilities", "Entertainment/Events", "Professional Services", "Small Office", "Warehouse/Distribution", "Other", ], label="Business Type", value="Retail Store", visible=False, interactive=True, info="Select your business type for specialized recommendations", ) nl_vulnerable_groups = gr.CheckboxGroup( choices=[ "Elderly", "Children", "Chronic Health Conditions", "Pregnant", ], label="Vulnerable Groups in Household", ) analyze_btn = gr.Button( "🔍 Analyze Query & Get Recommendations", variant="primary", size="lg", ) with gr.Row(): gr.HTML("""

🛰️ Agentic Logs

""") with gr.Row(): nl_logs_box = gr.Textbox( value=logcatcher.get_logs(), label="Logs", lines=17, max_lines=25, interactive=False, elem_id="nl_terminal_logs", show_copy_button=True, container=False, ) nl_logs_timer = gr.Timer(0.5) nl_logs_timer.tick(get_logs, None, nl_logs_box) with gr.Row(): nl_location_map = gr.HTML( "
Map will appear here after analysis.
", label="Interactive Risk Map", ) # Résultats d'analyse en langage naturel dans un cadre custom (CSS) with gr.Row(): risk_analysis_output = gr.Markdown( "Enter your question above to get started.", label="Risk Analysis", elem_id="nl_risk_box", ) # Recommandations NL dans un cadre custom (CSS) with gr.Row(): recommendations_output = gr.Markdown( "Personalized recommendations will appear here.", label="AI-Generated Recommendations", elem_id="nl_rec_box", ) with gr.TabItem("⚙️ Settings"): with gr.Row(): with gr.Column(): gr.Markdown(""" ## 🔐 API Configuration Configure your API keys here to enable full functionality of CAVA-AI. """) # Current API status api_status_display = gr.Markdown( self.get_current_api_status(), label="Current API Status", elem_id="api_status_box" ) # API Key inputs anthropic_api_input = gr.Textbox( label="Anthropic API Key", placeholder="sk-ant-...", type="password", info="Required for AI analysis. Get one at: https://console.anthropic.com/" ) nasa_api_input = gr.Textbox( label="NASA FIRMS API Key (Optional)", placeholder="Your NASA FIRMS Map Key", type="password", info="Optional for enhanced wildfire data. Get one at: https://firms.modaps.eosdis.nasa.gov/api/" ) # Update button update_keys_btn = gr.Button( "🔄 Update API Keys", variant="primary", size="lg" ) # Status message update_status = gr.Markdown( "", label="Update Status", elem_id="update_status_box" ) with gr.Column(): gr.Markdown(""" ## 📖 API Key Information ### Anthropic API Key (Required) - **Purpose**: Powers the AI agents for climate risk analysis - **Format**: Starts with `sk-ant-` - **Get Key**: [Anthropic Console](https://console.anthropic.com/) - **Pricing**: Pay per token usage ### NASA FIRMS API Key (Optional) - **Purpose**: Enhanced wildfire detection data - **Format**: Alphanumeric string - **Get Key**: [NASA FIRMS](https://firms.modaps.eosdis.nasa.gov/api/) - **Cost**: Free with registration ### 🔒 Security Notes - API keys are stored in environment variables - Keys are masked in the interface for security - Restart the application after updating keys - Never share your API keys publicly ### 🛠️ Troubleshooting - Ensure API keys are valid and active - Check your Anthropic account billing status - Verify network connectivity for API calls """) # Connect the update button update_keys_btn.click( fn=self.validate_and_update_api_key, inputs=[anthropic_api_input, nasa_api_input], outputs=[update_status] ) # Refresh status when tab is accessed update_keys_btn.click( fn=self.get_current_api_status, inputs=[], outputs=[api_status_display] ) # CSS pour les cadres custom gr.HTML(""" """) profile_dropdown.change( fn=self.update_business_visibility, inputs=[profile_dropdown], outputs=[business_type_dropdown], ) nl_profile_type.change( fn=self.update_business_visibility, inputs=[nl_profile_type], outputs=[nl_business_type_dropdown], ) country_dropdown.change( fn=self.update_cities, inputs=[country_dropdown], outputs=[city_suggestions, state_dropdown, location_map], ) city_input.change( fn=self.update_map_from_location, inputs=[country_dropdown, city_input, state_dropdown], outputs=[location_status, location_map], ) analyze_location_btn.click( fn=self.analyze_with_dropdown, inputs=[ country_dropdown, city_input, state_dropdown, profile_dropdown, business_type_dropdown, vulnerable_groups, ], outputs=[dropdown_risk_summary, dropdown_recommendations, location_map], show_progress="full", ) analyze_btn.click( fn=self.analyze_user_input, inputs=[ user_query, nl_profile_type, nl_business_type_dropdown, nl_vulnerable_groups, ], outputs=[ risk_analysis_output, recommendations_output, nl_location_map, ], show_progress="full", ) return app