import gradio as gr from datetime import datetime import sys import threading 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 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() == "farmer/agriculture": profile_context = " Emphasize agricultural risks including crop threats, soil conditions, water availability, extreme weather impacts on farming operations, and seasonal climate patterns." 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="πŸ›°οΈ SentinelO – Climate Risk Evaluation MultiAgents" ) as app: gr.Markdown( """ # πŸ›°οΈ SentinelO – Climate Risk Evaluation MultiAgents
πŸ€– What does SentinelO do?

SentinelO's AI agents instantly analyze climate risks ( πŸŒͺ️ Weather, 🌊 Flood, 🌍 Earthquake, πŸ”₯ Wildfire, 🌫️ Air quality, πŸ“ˆ Climate trends, β˜€οΈ Solar radiation, 🌊 Marine forecast ) for any 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 SentinelO?
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="France", interactive=True, ) city_input = gr.Textbox( label="Enter City Name", placeholder="e.g., Bordeaux, Lyon, Marseille, ...", value="Lorient", 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("France"), visible=True ) with gr.Column(): profile_dropdown = gr.Dropdown( choices=[ "General Public", "Business Owner", "Farmer/Agriculture", "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 New York get flooded tomorrow if we don't win the Hackaton ?", 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 Los Angeles this week?" - "I live in Lorient (Bretagne), can I run outside this evening ?" - "I'm planning to move to Miami, what climate risks should I be aware of?" - "How should my farm in Iowa prepare for climate change?" - "What emergency preparations should my business in Tokyo make for earthquakes?" """ ) with gr.Column(scale=1): nl_profile_type = gr.Dropdown( choices=[ "General Public", "Business Owner", "Farmer/Agriculture", "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", ) # 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