cava-ai / ui /ui.py
frankthtank's picture
added api key tab
6292f56 verified
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.",
"",
"<div style='text-align: center; padding: 50px; background-color: #f0f0f0; border-radius: 10px;'>Map will appear here after analysis.</div>",
)
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
<div style='background: linear-gradient(90deg, #f6f8fa 0%, #e2eafc 100%); border-radius: 10px; padding: 16px 18px; font-size: 16px; margin-bottom: 10px;'>
<b>πŸ€– What does CAVA-AI do?</b>
<br><br>
CAVA-AI's AI agents instantly analyze climate risks <b>(
πŸŒͺ️ Weather,
🌊 Flood,
🌍 Earthquake,
πŸ”₯ Wildfire,
🌫️ Air quality,
πŸ“ˆ Climate trends,
β˜€οΈ Solar radiation,
🌊 Marine forecast
)</b> for a specified location, providing you with clear, actionable recommendations.
<br><br>
<i>Analysis is fully automated, always up to date, and based on leading data sources: OpenStreetMap πŸ—ΊοΈ, Open-Meteo 🌦️, USGS 🌎, NASA FIRMS πŸ”₯.</i>
<br><br>
<b>How to use CAVA-AI?</b><br>
Use the <b>quick location selection</b> (dropdowns and map) 🌍, or ask complex, personalized questions in <b>natural language</b> πŸ’¬.
</div>
"""
)
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("""
<div style="display: flex; align-items: center; gap: 10px;">
<h3 style="margin: 0;">πŸ›°οΈ Agentic Logs</h3>
</div>
""")
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("""
<div style="display: flex; align-items: center; gap: 10px;">
<h3 style="margin: 0;">πŸ›°οΈ Agentic Logs</h3>
</div>
""")
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(
"<div style='text-align: center; padding: 50px; background-color: #f0f0f0; border-radius: 10px;'>Map will appear here after analysis.</div>",
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("""
<style>
#risk_summary_box, #recommendations_box, #nl_risk_box, #nl_rec_box, #api_status_box, #update_status_box {
border: 2px solid #007aff;
border-radius: 13px;
background: #fafdff;
box-shadow: 0 2px 12px rgba(80,140,255,0.08);
padding: 20px 15px;
margin-top: 10px;
margin-bottom: 18px;
}
#terminal_logs textarea, #nl_terminal_logs textarea {
background-color: #181a1b !important;
color: #00ff66 !important;
font-family: 'Fira Mono', 'Consolas', monospace !important;
font-size: 15px;
border-radius: 9px !important;
border: 2px solid #31343a !important;
box-shadow: 0 2px 6px rgba(0,0,0,0.19);
padding: 12px 10px !important;
min-height: 320px !important;
max-height: 420px !important;
letter-spacing: 0.5px;
line-height: 1.5;
overflow-y: auto !important;
resize: vertical !important;
scrollbar-width: thin;
scrollbar-color: #6cf97c #282c34;
}
#terminal_logs, #nl_terminal_logs {
width: 100% !important;
}
</style>
""")
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