Earthquakes (as of January 26, 2025) — Source: USGS.gov
Xtreme Weather App is an advanced disaster preparedness multi-agent system built with LangChain and Gemini-2.5-pro-preview-05–06 that provides personalized emergency guidance using a Streamlit interface. The system processes user location, fetches comprehensive weather data, identifies nearby emergency resources, and transforms technical information into clear, actionable advice to help users prepare for potential climate threats in their area.
The code I will provide implements this extreme weather monitoring and alert system that combines multiple weather and disaster-related data sources. The application takes an user’s address as input and converts it to latitude/longitude coordinates using the Google Maps Geocoding API. These coordinates are then used to query various APIs including OpenWeatherMap for weather data, USGS for earthquake information, NOAA for tsunami warnings, Places API, and additional endpoints for floods and hurricane alerts.
The app uses Google’s gemini-2.5-pro-preview-05–06 model in two ways: first through a DisasterAdvisorAgent that helps process and route queries, and second via an ExplanationAgent via Vertex AI that generates detailed natural language analyses of the collected weather and hazard data. The Gemini integration allows for intelligent processing of the data and generation of human-readable summaries and recommendations.
The application interface is built using Streamlit and features several interactive components. The main dashboard displays current weather metrics in a three-column layout showing temperature, humidity, and wind conditions. Below this dashboard are expandable sections for the 5-day forecast, seismic activity reports, active alerts (hurricanes, tsunamis), and emergency resources, as you will see ahead. The interface includes an interactive map showing nearby emergency facilities like hospitals, police stations, and shelters, with each location having its own expandable card containing detailed information and location in a map.
This application serves a critical purpose in disaster preparedness and response by aggregating multiple threat vectors (weather, seismic, volcanic, tsunami, floods) into a single dashboard. Users can quickly assess their risk level for various natural disasters, find nearby emergency resources, and receive AI-generated recommendations for safety measures, including available shelters nearby. The integration of multiple data sources and AI-powered analysis helps users make informed decisions during potential emergency situations, making it particularly valuable for areas prone to natural disasters. I will provide the full code of this solution.
Let’s start with the structure of the project:
Project structure
Here, the main notebook is app.py. Dockerfile and requirements.txt are used for deployment in Google Cloud Run. In the folder components there is a file called weather_dashboard.py that will integrate the React component weather_dashboard.jsx into Streamlit for cool visuals.
Let’s get our environment ready. Here’s our requirements.txt file:
uvicorn==0.24.0
pydantic==2.9.2
google-cloud-aiplatform==1.25.0
langchain
langchain-core
langchain-google-genai==2.0.9
google-generativeai==0.8.4
python-dotenv==1.0.0
google-auth==2.23.0
google-genai==0.6.0
Markdown==3.7
ipython==8.18.1
streamlit==1.41.1
extra-streamlit-components==0.1.71
Now, let’s create a Python environment, activate it and install the necessary libraries:
python3 -m venv streamlit-env
. streamlit-env/bin/activate
pip install -r requirements.txt
Now, authorize the application in Google Cloud and set the project you will use:
gcloud auth application-default login
gcloud config set project your-project
Let’s start with the main app.py file. We do the necessary imports:
import os
from typing import Dict, List, Any
from langchain.agents import Tool
from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain
from langchain.tools import BaseTool
from langchain.agents import AgentType, initialize_agent
from pydantic import BaseModel, Field
from langchain_google_genai import ChatGoogleGenerativeAI
import requests
import json
from datetime import datetime, timedelta
import re
import math
from google.cloud import secretmanager
import pandas as pd
from components.weather_dashboard import render_weather_dashboard
from google import genai
from google.genai import types
import base64
import markdown
from IPython.display import display, Markdown, HTML
import streamlit as st
import streamlit.components.v1 as components
Now you will need some API Keys:
- Openweather API Key
- Google Maps API Key
- Google Gemini API Key and also Vertex AI Key
- NOAA Token
The Openweather, USGS and NOAA API keys are free, some only need registration and a credit card for more than 60 requests per minute. Google API Keys are paid, of course, and in this tutorial, for each 1.00 USD spent in Google Geocoding API, there will be a cost of 29.36 USD in Google Places API.
Put these keys as secrets inside Google Cloud Secret Manager, in order to secure your code.
Then, get your project number by running:
gcloud projects describe $PROJECT_ID --format="value(projectNumber)"
Make your code retrieve the secrets from Secret Manager, so that they can be used without exposing hardcoded credentials:
project_id='1234567890'
def get_secret(project_id: str, secret_id: str, version_id: str = "1") -> str:
client = secretmanager.SecretManagerServiceClient()
name = f"projects/{project_id}/secrets/{secret_id}/versions/{version_id}"
response = client.access_secret_version(request={"name": name})
return response.payload.data.decode('UTF-8')
OPENWEATHER_API_KEY = get_secret(project_id, 'OPENWEATHER_API_KEY')
GOOGLE_MAPS_API_KEY = get_secret(project_id, 'GOOGLE_MAPS_API_KEY')
GOOGLE_AI_API_KEY = get_secret(project_id, 'GOOGLE_AI_API_KEY')
NOAA_TOKEN = get_secret(project_id, 'NOAA_TOKEN')
Now, let’s define the wrapper for Gemini API calls_,_ a function to call gemini-2.5-pro-preview-05–06:
def generate(prompt):
client = genai.Client(
vertexai=True,
project="your-project",
location="us-central1"
)
model = "gemini-2.5-pro-preview-05-06"
contents = [
types.Content(
role="user",
parts=[
types.Part.from_text(prompt)
]
),
]
generate_content_config = types.GenerateContentConfig(
temperature = 1,
top_p = 0.95,
max_output_tokens = 512,
response_modalities = ["TEXT"],
safety_settings = [types.SafetySetting(
category="HARM_CATEGORY_HATE_SPEECH",
threshold="OFF"
),types.SafetySetting(
category="HARM_CATEGORY_DANGEROUS_CONTENT",
threshold="OFF"
),types.SafetySetting(
category="HARM_CATEGORY_SEXUALLY_EXPLICIT",
threshold="OFF"
),types.SafetySetting(
category="HARM_CATEGORY_HARASSMENT",
threshold="OFF"
)],
)
for chunk in client.models.generate_content_stream(
model = model,
contents = contents,
config = generate_content_config,
):
print(chunk.text, end="")
Add these Pydantic data models that define structured data types:
class LocationInfo(BaseModel):
address: str = Field(description="User's address")
lat: float = Field(description="Latitude")
lng: float = Field(description="Longitude")
class WeatherInfo(BaseModel):
current_conditions: dict
forecast: List[dict]
alerts: List[dict]
Now, let’s set up our Google Geocoding tool. This tool will get latitude and longitude coordinates for a given address:
class GeocodingTool(BaseTool):
name: str = Field(default="geocoding_tool")
description: str = Field(default="Get latitude and longitude coordinates for a given address")
return_direct: bool = Field(default=False)
def _run(self, address: str) -> Dict:
url = f"https://maps.googleapis.com/maps/api/geocode/json?address={address}&key={GOOGLE_MAPS_API_KEY}"
response = requests.get(url)
data = response.json()
if data['status'] == 'OK':
location = data['results'][0]['geometry']['location']
return {
"address": data['results'][0]['formatted_address'],
"lat": location['lat'],
"lng": location['lng']
}
else:
raise Exception(f"Geocoding failed: {data['status']}")
async def _arun(self, address: str) -> Dict:
raise NotImplementedError("Async not implemented")
We will also implement our Weather Tool , that is a primary class that:
- Gets current weather and forecasts from OpenWeatherMap
- Fetches earthquake data from USGS
- Retrieves tsunami warnings from NOAA
- Gets hurricane alerts from National Weather Service
- Checks volcanic activity from USGS
- Includes helper methods for each data type
class WeatherTool(BaseTool):
name: str = Field(default="weather_tool")
description: str = Field(default="Get weather information and alerts for a specific location")
return_direct: bool = Field(default=False)
def _run(self, location_dict: Any) -> Dict:
try:
if isinstance(location_dict, str):
location_dict = eval(location_dict)
# Get basic weather data
weather_info = self._get_weather_data(location_dict)
# Add earthquake data
earthquake_data = self._get_earthquake_data(location_dict)
weather_info["seismic_activity"] = earthquake_data
# Add hurricane data
hurricane_data = self._get_hurricane_data(location_dict)
weather_info["hurricane_alerts"] = hurricane_data
# Add tsunami data
tsunami_data = self._get_tsunami_data(location_dict)
weather_info["tsunami_alerts"] = tsunami_data
volcano_data = self._get_volcano_data(location_dict)
weather_info["volcano_activity"] = volcano_data
return weather_info
except Exception as e:
print(f"Debug - Error in WeatherTool: {str(e)}")
print(f"Debug - Error type: {type(e)}")
import traceback
print(f"Debug - Traceback: {traceback.format_exc()}")
raise Exception(f"Weather data fetch failed: {str(e)}")
def _get_volcano_data(self, location_dict: Dict) -> List[Dict]:
"""Fetch volcano data from USGS Earthquake API filtering for volcanic events"""
try:
# Calculate dates for the query (last 10 days)
from datetime import datetime, timedelta
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=10)
# Format dates in YYYY-MM-DD format
start_str = start_date.strftime("%Y-%m-%d")
end_str = end_date.strftime("%Y-%m-%d")
# Construct the URL with query parameters
url = f"https://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson&starttime={start_str}&endtime={end_str}&eventtype=volcanic%20eruption"
headers = {
'User-Agent': 'DisasterAdvisor/1.0',
'Accept': 'application/json'
}
response = requests.get(url, headers=headers, timeout=10)
if response.status_code != 200:
print(f"Volcano API returned status code: {response.status_code}")
print(f"Response content: {response.text}")
return []
data = response.json()
nearby_volcanoes = []
# Process features from the GeoJSON
for feature in data.get('features', []):
properties = feature.get('properties', {})
geometry = feature.get('geometry', {})
if geometry and geometry.get('type') == 'Point':
coordinates = geometry.get('coordinates', [0, 0])
nearby_volcanoes.append({
"name": properties.get('place', 'Unknown Location'),
"type": "Volcanic Activity",
"status": "Active",
"alert_level": "Warning",
"magnitude": properties.get('mag'),
"time": datetime.fromtimestamp(properties.get('time', 0)/1000).isoformat() if properties.get('time') else None,
"details": properties.get('detail', 'Volcanic activity detected')
})
return nearby_volcanoes
except requests.RequestException as e:
print(f"Network error fetching volcano data: {str(e)}")
return []
except ValueError as e:
print(f"Error parsing volcano data: {str(e)}")
return []
except Exception as e:
print(f"Unexpected error fetching volcano data: {str(e)}")
print(f"Error details: {str(e. __class__. __name__ )}")
import traceback
print(f"Traceback: {traceback.format_exc()}")
return []
def _get_weather_data(self, location_dict: Dict) -> Dict:
"""Get weather information from OpenWeatherMap"""
lat = float(location_dict["lat"])
lng = float(location_dict["lng"])
# Current weather
current_url = f"https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lng}&appid={OPENWEATHER_API_KEY}&units=metric"
current_response = requests.get(current_url)
if current_response.status_code != 200:
raise Exception(f"Current weather API failed with status {current_response.status_code}")
current_data = current_response.json()
# 5-day forecast
forecast_url = f"https://api.openweathermap.org/data/2.5/forecast?lat={lat}&lon={lng}&appid={OPENWEATHER_API_KEY}&units=metric"
forecast_response = requests.get(forecast_url)
if forecast_response.status_code != 200:
raise Exception(f"Forecast API failed with status {forecast_response.status_code}")
forecast_data = forecast_response.json()
# Process forecast data
forecast_list = []
if isinstance(forecast_data, dict) and 'list' in forecast_data:
forecast_list = forecast_data['list']
# Structure the complete response
weather_data = {
"current_conditions": {
"temperature": current_data.get('main', {}).get('temp'),
"weather": current_data.get('weather', [{}])[0].get('description'),
"humidity": current_data.get('main', {}).get('humidity'),
"wind_speed": current_data.get('wind', {}).get('speed'),
"wind_direction": current_data.get('wind', {}).get('deg'),
"pressure": current_data.get('main', {}).get('pressure'),
"visibility": current_data.get('visibility'),
"feels_like": current_data.get('main', {}).get('feels_like')
},
"forecast": [
{
"datetime": item.get('dt_txt'),
"temperature": item.get('main', {}).get('temp'),
"weather": item.get('weather', [{}])[0].get('description'),
"humidity": item.get('main', {}).get('humidity'),
"wind_speed": item.get('wind', {}).get('speed'),
"wind_direction": item.get('wind', {}).get('deg'),
"pressure": item.get('main', {}).get('pressure'),
"feels_like": item.get('main', {}).get('feels_like')
}
for item in forecast_list[:5] # Get next 5 timestamps
],
"alerts": [] # Will be populated by other methods
}
return weather_data
def _get_earthquake_data(self, location_dict: Dict) -> List[Dict]:
"""Fetch recent earthquake data from USGS"""
lat = location_dict["lat"]
lng = location_dict["lng"]
# USGS API endpoint for earthquakes within 300km in the past 7 days
url = f"https://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson&latitude={lat}&longitude={lng}&maxradiuskm=300&minmagnitude=2.5&orderby=time"
response = requests.get(url)
data = response.json()
earthquakes = []
for feature in data["features"]:
earthquakes.append({
"magnitude": feature["properties"]["mag"],
"location": feature["properties"]["place"],
"time": datetime.fromtimestamp(feature["properties"]["time"] / 1000.0).isoformat(),
"url": feature["properties"]["url"]
})
return earthquakes
def _get_hurricane_data(self, location_dict: Dict) -> List[Dict]:
"""Fetch hurricane warnings from National Weather Service API"""
lat = location_dict["lat"]
lng = location_dict["lng"]
# NWS API endpoint
headers = {
"Accept": "application/geo+json",
"User-Agent": "(disaster-advisor-app.com, contact@disaster-advisor-app.com)"
}
try:
# Get active alerts for the area
alerts_url = f"https://api.weather.gov/alerts/active?point={lat},{lng}"
alerts_response = requests.get(alerts_url, headers=headers)
alerts_data = alerts_response.json()
# Filter for hurricane-related alerts
hurricane_alerts = []
hurricane_terms = ['hurricane', 'tropical storm', 'tropical cyclone']
for feature in alerts_data.get('features', []):
properties = feature.get('properties', {})
event = properties.get('event', '').lower()
if any(term in event for term in hurricane_terms):
hurricane_alerts.append({
"event": properties.get('event'),
"severity": properties.get('severity'),
"headline": properties.get('headline'),
"description": properties.get('description'),
"instruction": properties.get('instruction'),
"onset": properties.get('onset'),
"expires": properties.get('expires')
})
return hurricane_alerts
except Exception as e:
print(f"Error fetching hurricane data: {str(e)}")
return []
def _get_tsunami_data(self, location_dict: Dict) -> List[Dict]:
"""Fetch tsunami warnings from NOAA's Tsunami Warning System"""
lat = location_dict["lat"]
lng = location_dict["lng"]
# NOAA Tsunami Warning Center API
# Using the CAP (Common Alerting Protocol) feed
url = "https://www.tsunami.gov/events/xml/PAAQAtom.xml"
headers = {
"User-Agent": "(disaster-advisor-app.com, contact@disaster-advisor-app.com)"
}
try:
response = requests.get(url, headers=headers)
# The feed is in XML format
from xml.etree import ElementTree
root = ElementTree.fromstring(response.content)
# Parse tsunami alerts
tsunami_alerts = []
# XML namespaces used in the feed
namespaces = {
'cap': 'urn:oasis:names:tc:emergency:cap:1.2',
'atom': 'http://www.w3.org/2005/Atom'
}
for entry in root.findall('.//atom:entry', namespaces):
# Get the CAP alert
cap_alert = entry.find('.//cap:alert', namespaces)
if cap_alert is not None:
info = cap_alert.find('.//cap:info', namespaces)
if info is not None:
# Check if this alert affects our location
area = info.find('.//cap:area', namespaces)
if area is not None:
# Convert the area description to a rough bounding box
# and check if our location falls within it
if self._location_in_alert_area(lat, lng, area):
tsunami_alerts.append({
"event": info.find('.//cap:event', namespaces).text if info.find('.//cap:event', namespaces) is not None else None,
"severity": info.find('.//cap:severity', namespaces).text if info.find('.//cap:severity', namespaces) is not None else None,
"urgency": info.find('.//cap:urgency', namespaces).text if info.find('.//cap:urgency', namespaces) is not None else None,
"description": info.find('.//cap:description', namespaces).text if info.find('.//cap:description', namespaces) is not None else None,
"instruction": info.find('.//cap:instruction', namespaces).text if info.find('.//cap:instruction', namespaces) is not None else None,
"effective": info.find('.//cap:effective', namespaces).text if info.find('.//cap:effective', namespaces) is not None else None,
"expires": info.find('.//cap:expires', namespaces).text if info.find('.//cap:expires', namespaces) is not None else None
})
return tsunami_alerts
except Exception as e:
print(f"Error fetching tsunami data: {str(e)}")
return []
def _location_in_alert_area(self, lat: float, lng: float, area_element) -> bool:
"""Helper function to determine if a location falls within a CAP alert area"""
area_desc = area_element.find('./cap:areaDesc', {'cap': 'urn:oasis:names:tc:emergency:cap:1.2'})
if area_desc is not None:
return True
return False
async def _arun(self, location: Dict) -> Dict:
raise NotImplementedError("Async not implemented")
Now we set up our EmergencyResourcesTool , that finds nearby hospitals, police stations, fire stations, and shelters using Google Places API.
class EmergencyResourcesTool(BaseTool):
name: str = Field(default="emergency_resources_tool")
description: str = Field(default="Find nearby emergency resources and shelters")
return_direct: bool = Field(default=False)
def _run(self, location_dict: Dict) -> List[Dict]:
try:
if isinstance(location_dict, str):
location_dict = eval(location_dict)
lat = location_dict["lat"]
lng = location_dict["lng"]
place_types = ["hospital", "police", "fire_station"]
all_resources = []
for place_type in place_types:
url = f"https://maps.googleapis.com/maps/api/place/nearbysearch/json?location={lat},{lng}&radius=5000&type={place_type}&key={GOOGLE_MAPS_API_KEY}"
response = requests.get(url)
data = response.json()
if data['status'] == 'OK':
for place in data['results']:
all_resources.append({
'name': place['name'],
'address': place.get('vicinity', ''),
'location': place['geometry']['location'],
'type': place_type
})
# Additionally search for emergency shelters using keyword
shelter_url = f"https://maps.googleapis.com/maps/api/place/nearbysearch/json?location={lat},{lng}&radius=5000&keyword=emergency+shelter&key={GOOGLE_MAPS_API_KEY}"
shelter_response = requests.get(shelter_url)
shelter_data = shelter_response.json()
if shelter_data['status'] == 'OK':
for place in shelter_data['results']:
all_resources.append({
'name': place['name'],
'address': place.get('vicinity', ''),
'location': place['geometry']['location'],
'type': 'shelter'
})
return all_resources
except Exception as e:
print(f"Error in emergency resources fetch: {str(e)}")
raise Exception(f"Emergency resources fetch failed: {str(e)}")
async def _arun(self, location: Dict) -> List[Dict]:
raise NotImplementedError("Async not implemented")
Now we will build the Langchain agents. The DisasterAdvisorAgent has the following abilities:
- Coordinates tools using Google’s Gemini
- Processes user queries
- Routes requests to appropriate tools
- Maintains conversation memory
- Has geocoding, weather and emergency resources tools
class DisasterAdvisorAgent:
def __init__ (self):
self.llm = ChatGoogleGenerativeAI(model="gemini-2.5-pro-preview-05-06", google_api_key=GOOGLE_AI_API_KEY)
self.memory = ConversationBufferMemory(memory_key="chat_history")
# Initialize tools
self.geocoding_tool = GeocodingTool()
self.weather_tool = WeatherTool()
self.emergency_resources_tool = EmergencyResourcesTool()
self.tools = [
Tool(
name="Geocoding",
func=self.geocoding_tool._run,
description="Convert address to coordinates. Input should be a string address."
),
Tool(
name="Weather",
func=self.weather_tool._run,
description="Get weather information and alerts for a location. Input should be the direct output from the Geocoding tool."
),
Tool(
name="Emergency Resources",
func=self.emergency_resources_tool._run,
description="Find nearby emergency resources. Input should be a dictionary containing 'lat' and 'lng' keys."
)
]
# Initialize the agent
self.agent_executor = initialize_agent(
tools=self.tools,
llm=self.llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
memory=self.memory,
verbose=True,
handle_parsing_errors=True
)
def get_response(self, user_input: str) -> str:
"""
Process user input and return a response
"""
try:
if "climate threats" in user_input.lower() or "weather" in user_input.lower():
# First get location data
location_response = self.geocoding_tool._run(user_input.split("address is ")[-1].strip())
import time
time.sleep(2)
# Then get weather data using the location
weather_data = self.weather_tool._run(location_response)
# Add emergency resources data
try:
emergency_resources = self.emergency_resources_tool._run(location_response)
weather_data['emergency_resources'] = emergency_resources
except Exception as e:
print(f"Error fetching emergency resources: {str(e)}")
weather_data['emergency_resources'] = []
# Return the combined data
return weather_data
else:
# For non-weather queries, use the normal agent response
response = self.agent_executor.run(user_input)
return response
except Exception as e:
return f"An error occurred: {str(e)}"
And our ExplanationAgent , that runs in Google Cloud VertexAI, that:
- Uses Gemini to generate natural language analysis of weather data
- Creates markdown-formatted reports
- Provides safety recommendations
class ExplanationAgent:
def __init__ (self):
self.client = genai.Client(
vertexai=True,
project="your-project",
location="us-central1"
)
self.generate_content_config = types.GenerateContentConfig(
temperature=1,
top_p=0.95,
max_output_tokens=8192,
response_modalities=["TEXT"],
safety_settings=[
types.SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="OFF"),
types.SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="OFF"),
types.SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="OFF"),
types.SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="OFF")
]
)
def explain_weather_data(self, weather_data: dict, location: str):
# Extract data with safe fallbacks
current = weather_data.get('current_conditions', {})
forecast = weather_data.get('forecast', [])
seismic = weather_data.get('seismic_activity', [])
tsunamis = weather_data.get('tsunami_alerts', [])
volcanoes = weather_data.get('volcano_activity', [])
hurricanes = weather_data.get('hurricane_alerts', [])
floods = weather_data.get('flood_alerts', [])
emergency_resources = weather_data.get('emergency_resources', [])
# Format emergency resources information
emergency_info = ""
if emergency_resources:
emergency_info = "Nearby Emergency Resources:\n"
resources_by_type = {}
# Group resources by type
for resource in emergency_resources:
resource_type = resource.get('type', 'other')
if resource_type not in resources_by_type:
resources_by_type[resource_type] = []
resources_by_type[resource_type].append(resource)
# Format each type of resource
for resource_type, resources in resources_by_type.items():
emergency_info += f"\n{resource_type.upper()}:\n"
for resource in resources:
emergency_info += f"""
- Name: {resource.get('name')}
Address: {resource.get('address')}
Distance: {resource.get('distance', 'N/A')}
"""
else:
emergency_info = "No emergency resource information available."
# Format the seismic activity data for better readability
seismic_info = ""
if seismic:
seismic_info = "Recent earthquakes:\n"
for quake in seismic:
seismic_info += f"""
- Magnitude: {quake.get('magnitude')}
Location: {quake.get('location')}
Time: {quake.get('time')}
More info: {quake.get('url')}
"""
else:
seismic_info = "No recent seismic activity reported."
tsunami_info = ""
if tsunamis:
tsunami_info = "Active tsunami alerts:\n"
for alert in tsunamis:
tsunami_info += f"""
- Event: {alert.get('event')}
Severity: {alert.get('severity')}
Status: {alert.get('status')}
Expected Time: {alert.get('expected_time')}
Affected Areas: {alert.get('affected_areas')}
Instructions: {alert.get('instructions')}
"""
else:
tsunami_info = "No active tsunami alerts reported."
hurricane_info = ""
if hurricanes:
hurricane_info = "Active hurricane alerts:\n"
for alert in hurricanes:
hurricane_info += f"""
- Event: {alert.get('event')}
Severity: {alert.get('severity')}
Headline: {alert.get('headline')}
Description: {alert.get('description')}
Instruction: {alert.get('instruction')}
Onset: {alert.get('onset')}
Expires: {alert.get('expires')}
"""
else:
hurricane_info = "No active hurricane alerts reported."
volcano_info = ""
if volcanoes:
volcano_info = "Active volcanoes in the area:\n"
for volcano in volcanoes:
volcano_info += f"""
- Name: {volcano.get('name')}
Type: {volcano.get('type')}
Status: {volcano.get('status')}
Alert Level: {volcano.get('alert_level')}
Distance: {volcano.get('distance_km', 0):.1f} km
Last Eruption: {volcano.get('last_eruption')}
Activity Details: {volcano.get('details')}
"""
else:
volcano_info = "No active volcanoes reported in the area."
# Format hurricane information with enhanced details
hurricane_info = ""
if hurricanes:
hurricane_info = "Active hurricane/cyclone alerts:\n"
for alert in hurricanes:
hurricane_info += f"""
- Source: {alert.get('source')}
Event: {alert.get('event')}
Severity: {alert.get('severity')}
Distance: {alert.get('distance_km', 'Unknown')} km
Wind Speed: {alert.get('wind_speed', 'Unknown')}
Pressure: {alert.get('pressure', 'Unknown')}
Movement: {alert.get('movement', 'Unknown')}
Details: {alert.get('details')}
"""
else:
hurricane_info = "No active hurricane/cyclone alerts reported."
# Format flood information
flood_info = ""
if floods:
flood_info = "Active flood alerts:\n"
for alert in floods:
flood_info += f"""
- Source: {alert.get('source')}
Event Type: {alert.get('event_type', 'Flood')}
Severity: {alert.get('severity')}
Status: {alert.get('status')}
Affected Area: {alert.get('affected_area')}
Start Date: {alert.get('start_date')}
Forecast: {alert.get('forecast', 'Not available')}
"""
else:
flood_info = "No active flood alerts reported."
prompt = f"""
Analyze the following weather and emergency data for {location}:
CURRENT CONDITIONS:
Temperature: {current.get('temperature')}°C
Weather: {current.get('weather')}
Humidity: {current.get('humidity')}%
Wind Speed: {current.get('wind_speed')} m/s
Wind Direction: {current.get('wind_direction')}°
Pressure: {current.get('pressure')} hPa
Visibility: {current.get('visibility')} m
Feels Like: {current.get('feels_like')}°C
FORECAST:
{json.dumps(forecast, indent=2)}
SEISMIC ACTIVITY:
{seismic_info}
TSUNAMI ALERTS:
{tsunami_info}
VOLCANIC ALERTS:
{volcano_info}
HURRICANE/CYCLONE ALERTS:
{hurricane_info}
FLOOD ALERTS:
{flood_info}
Please provide a concise analysis formatted in markdown:
# ⚠️ Emergency Status Summary
[Overview of immediate risks]
# Current Conditions
- Highlight anomalies in temperature, humidity, wind conditions
- Highlight any severe weather conditions
- Highlight any immediate concerns
# 📈 Seismic Activity
- List all recent earthquakes with magnitude and location
- Evaluate potential aftershock risks
- Note proximity to populated areas
# 🌊 Tsunami Alerts
- List tsunamis, if applicable
- Note areas at risk
- Include evacuation instructions if provided
# 🌋 Volcanic Alerts
- List any active volcanoes in the area
- Note current alert levels and activity status
- Include distance from location and potential risks
- Highlight any significant recent changes in activity
# 🌀 Hurricane/Cyclone Alerts
- List any active storms
- Note severity, wind speeds, and movement patterns
- Include specific threat levels for the location
- Highlight expected timeline and progression
# 🌊 Flood Alerts
- List active flood warnings
- Note severity and affected areas
- Include water levels and forecasts if available
- Highlight areas at immediate risk
# ⛑️ Safety Recommendations
1. [Immediate actions needed]
2. [Preparation steps]
3. [Emergency supplies if needed]
4. [Evacuation considerations if relevant]
Include all numerical data where available and be specific about potential risks.
Prioritize immediate threats and provide clear, actionable guidance.
"""
contents = [
types.Content(
role="user",
parts=[types.Part.from_text(prompt)]
)
]
return self.client.models.generate_content_stream(
model="gemini-2.5-pro-preview-05-06",
contents=contents,
config=self.generate_content_config
)
Finally, we have our main() function, that:
- Creates the Streamlit interface
- Creates input form for address
- Displays weather dashboard
- Shows emergency resources
- Renders maps
- Displays AI-generated analysis
def main():
main_container = st.container()
with main_container:
st.title("🌪️ Xtreme Weather App")
st.write("")
st.markdown(" **Your Gemini multi-agent app for extreme events: hurricanes, earthquakes and tsunamis**")
st.write("")
st.write("Try any address or enter:")
st.write("Datah Village in Bali")
# Initialize agents
if 'disaster_advisor' not in st.session_state:
st.session_state.disaster_advisor = DisasterAdvisorAgent()
if 'explanation_agent' not in st.session_state:
st.session_state.explanation_agent = ExplanationAgent()
# Create columns for better layout
col1, col2 = st.columns([2, 1])
with col1:
address = st.text_input("📍 Enter your address:",
placeholder="e.g., 123 Main St, City, Country",
key="address_input")
with col2:
st.write("") # Add some spacing
st.write("") # Add some spacing
analyze_button = st.button("🔍 Get Analysis", type="primary")
if analyze_button and address:
with st.spinner("📊 Our agents are analyzing weather and emergency data..."):
try:
# Get weather data
response = st.session_state.disaster_advisor.get_response(
f"What are the extreme climate threats in my area for the next week? My address is {address}"
)
# Convert response to structured data
try:
# If response is a string that looks like a dict
if isinstance(response, str) and '{' in response:
import json
# Clean up the string if needed (remove any escape characters)
cleaned_response = response.replace('\n', '').replace('\\', '')
weather_data = json.loads(cleaned_response)
# If response is already a dict
elif isinstance(response, dict):
weather_data = response
else:
# Create a basic structure for text responses
weather_data = {
"current_conditions": {
"weather": response
}
}
# Display the weather dashboard
weather_dashboard_container = st.container()
with weather_dashboard_container:
st.markdown("### Current Weather Dashboard")
# Create three columns for current conditions
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Temperature", f"{weather_data['current_conditions']['temperature']}°C",
f"Feels like {weather_data['current_conditions']['feels_like']}°C")
with col2:
st.metric("Humidity", f"{weather_data['current_conditions']['humidity']}%")
with col3:
st.metric("Wind", f"{weather_data['current_conditions']['wind_speed']} m/s",
f"Direction {weather_data['current_conditions']['wind_direction']}°")
# Current weather condition
st.info(f"Current weather: {weather_data['current_conditions']['weather']}")
# Forecast section
st.markdown("### 5-Day Forecast")
for forecast in weather_data['forecast']:
with st.expander(f"Forecast for {forecast['datetime']}"):
cols = st.columns(4)
with cols[0]:
st.metric("Temperature", f"{forecast['temperature']}°C")
with cols[1]:
st.metric("Humidity", f"{forecast['humidity']}%")
with cols[2]:
st.metric("Wind", f"{forecast['wind_speed']} m/s")
with cols[3]:
st.write("Conditions:", forecast['weather'])
# Seismic activity section
if weather_data['seismic_activity']:
st.markdown("### Recent Seismic Activity")
for quake in weather_data['seismic_activity']:
with st.expander(f"Magnitude {quake['magnitude']} - {quake['location']}"):
st.write(f"Time: {quake['time']}")
st.write(f"Location: {quake['location']}")
st.markdown(f"[More details]({quake['url']})")
# Hurricane activity section
if weather_data.get('hurricane_alerts') or weather_data.get('tsunami_alerts'):
st.markdown("### ⚠️ Active Alerts ⚠️")
if weather_data['hurricane_alerts']:
st.error("Hurricane Alerts")
for alert in weather_data['hurricane_alerts']:
st.write(alert)
if weather_data['tsunami_alerts']:
st.error("Tsunami Alerts")
for alert in weather_data['tsunami_alerts']:
st.write(alert)
# Add Emergency Resources section here
st.markdown("### 🚑 Emergency Resources")
# Create tabs for different types of emergency resources
resource_types = ['hospital', 'police', 'fire_station', 'shelter']
tabs = st.tabs([resource.replace('_', ' ').title() for resource in resource_types])
# Group resources by type
resources_by_type = {}
for resource in weather_data.get('emergency_resources', []):
resource_type = resource.get('type', 'other')
if resource_type not in resources_by_type:
resources_by_type[resource_type] = []
resources_by_type[resource_type].append(resource)
# Display resources in respective tabs
for tab, resource_type in zip(tabs, resource_types):
with tab:
resources = resources_by_type.get(resource_type, [])
if resources:
for resource in resources:
with st.expander(f"📍 {resource['name']}"):
st.write(f" **Address:** {resource['address']}")
if 'location' in resource:
st.write(f" **Coordinates:** Lat {resource['location']['lat']}, Lng {resource['location']['lng']}")
# Create a map for this resource
map_data = pd.DataFrame({
'lat': [resource['location']['lat']],
'lon': [resource['location']['lng']]
})
st.map(map_data)
else:
st.info(f"No {resource_type.replace('_', ' ')} facilities found nearby.")
# Add disclaimer
st.caption("⚠️ Emergency resource information is provided for reference only. In case of emergency, always call your local emergency number (e.g., 911 in the US).")
# Get explanation using the structured data
explanation_stream = st.session_state.explanation_agent.explain_weather_data(
weather_data,
address
)
# Alerts section
if weather_data.get('hurricane_alerts') or weather_data.get('tsunami_alerts'):
st.markdown("### ⚠️ Active Alerts ⚠️")
if weather_data['hurricane_alerts']:
st.error("Hurricane Alerts")
for alert in weather_data['hurricane_alerts']:
st.write(alert)
if weather_data['tsunami_alerts']:
st.error("Tsunami Alerts")
for alert in weather_data['tsunami_alerts']:
st.write(alert)
# Get explanation using the structured data
explanation_stream = st.session_state.explanation_agent.explain_weather_data(
weather_data,
address
)
st.markdown("### Detailed Analysis")
full_response = ""
for chunk in explanation_stream:
full_response += chunk.text
st.markdown(full_response)
except Exception as e:
st.warning(f"Could not parse weather data: {str(e)}")
st.text(f"Raw response: {response}")
except Exception as e:
print(e)
if __name__ == " __main__":
main()
We still have two files to go, weather_dashboard.py and weather_dashboard.jsx, let’s build them:
weather_dashboard.py:
import streamlit as components
import json
import streamlit as st
def render_weather_dashboard(weather_data):
"""
Render the Weather Dashboard React component in Streamlit
"""
try:
# Debug: Print the type and content of weather_data
st.write("Debug - Weather Data Type:", type(weather_data))
st.write("Debug - Weather Data Content:", weather_data)
# If weather_data is a string, try to parse it as JSON
if isinstance(weather_data, str):
try:
weather_data = json.loads(weather_data)
except json.JSONDecodeError as e:
st.error(f"Failed to parse weather data as JSON: {str(e)}")
return
# Ensure weather_data has the expected structure
if not isinstance(weather_data, dict):
st.error(f"Weather data must be a dictionary, got {type(weather_data)}")
return
# Create a properly structured weather data object
formatted_weather_data = {
"current_conditions": weather_data.get("current_conditions", {
"temperature": 0,
"weather": "Unknown",
"humidity": 0,
"wind_speed": 0,
"wind_direction": 0,
"pressure": 0,
"feels_like": 0
}),
"forecast": weather_data.get("forecast", []),
"seismic_activity": weather_data.get("seismic_activity", [])
}
# Convert to JSON for the React component
weather_json = json.dumps(formatted_weather_data)
# Inject the component
components.html(
f"""
<div id="weather-dashboard"></div>
<script>
window.weatherData = {weather_json};
</script>
""",
height=600
)
except Exception as e:
st.error(f"Error in render_weather_dashboard: {str(e)}")
st.error(f"Error type: {type(e)}")
import traceback
st.error(f"Traceback: {traceback.format_exc()}")
weather_dashboard.jsx (a React component for cool visuals):
import React from 'react';
import { Card, CardHeader, CardTitle, CardContent } from '@/components/ui/card';
import { Alert, AlertDescription, AlertTitle } from '@/components/ui/alert';
import { ThermometerSun, Wind, Droplets, Eye, ArrowUp, Scale } from 'lucide-react';
const WeatherDashboard = ({ weatherData }) => {
if (!weatherData) {
return <div className="p-4">Loading weather data...</div>;
}
const WeatherIcon = ({ condition }) => {
const iconMap = {
'clear sky': '☀️',
'few clouds': '🌤️',
'scattered clouds': '⛅',
'broken clouds': '☁️',
'shower rain': '🌧️',
'rain': '🌧️',
'thunderstorm': '⛈️',
'snow': '🌨️',
'mist': '🌫️',
'heavy intensity rain': '⛈️',
'light rain': '🌦️',
'overcast clouds': '☁️'
};
return <span className="text-2xl">{iconMap[condition.toLowerCase()] || '🌡️'}</span>;
};
const formatDateTime = (dateStr) => {
const date = new Date(dateStr);
return date.toLocaleString();
};
return (
<div className="space-y-4 p-4">
<Card>
<CardHeader>
<CardTitle className="flex items-center gap-2">
<WeatherIcon condition={weatherData.current_conditions.weather} />
Current Weather Conditions
</CardTitle>
</CardHeader>
<CardContent>
<div className="grid grid-cols-2 md:grid-cols-4 gap-4">
<div className="flex items-center gap-2">
<ThermometerSun className="text-blue-500" />
<div>
<div className="text-sm text-gray-500">Temperature</div>
<div className="font-semibold">{weatherData.current_conditions.temperature}°C</div>
<div className="text-xs text-gray-400">Feels like: {weatherData.current_conditions.feels_like}°C</div>
</div>
</div>
<div className="flex items-center gap-2">
<Wind className="text-blue-500" />
<div>
<div className="text-sm text-gray-500">Wind</div>
<div className="font-semibold">{weatherData.current_conditions.wind_speed} m/s</div>
<div className="text-xs text-gray-400">Direction: {weatherData.current_conditions.wind_direction}°</div>
</div>
</div>
<div className="flex items-center gap-2">
<Droplets className="text-blue-500" />
<div>
<div className="text-sm text-gray-500">Humidity</div>
<div className="font-semibold">{weatherData.current_conditions.humidity}%</div>
</div>
</div>
<div className="flex items-center gap-2">
<Scale className="text-blue-500" />
<div>
<div className="text-sm text-gray-500">Pressure</div>
<div className="font-semibold">{weatherData.current_conditions.pressure} hPa</div>
</div>
</div>
</div>
</CardContent>
</Card>
{weatherData.forecast && weatherData.forecast.length > 0 && (
<Card>
<CardHeader>
<CardTitle>5-Day Forecast</CardTitle>
</CardHeader>
<CardContent>
<div className="space-y-4">
{weatherData.forecast.map((day, index) => (
<div key={index} className="flex items-center gap-4 p-2 hover:bg-gray-50 rounded">
<WeatherIcon condition={day.weather} />
<div className="flex-1">
<div className="font-semibold">{formatDateTime(day.datetime)}</div>
<div className="text-sm text-gray-500">{day.weather}</div>
</div>
<div className="text-right">
<div className="font-semibold">{day.temperature}°C</div>
<div className="text-sm text-gray-500">Humidity: {day.humidity}%</div>
</div>
</div>
))}
</div>
</CardContent>
</Card>
)}
{weatherData.seismic_activity && weatherData.seismic_activity.length > 0 && (
<Card>
<CardHeader>
<CardTitle>Recent Seismic Activity</CardTitle>
</CardHeader>
<CardContent>
<div className="space-y-4">
{weatherData.seismic_activity.map((event, index) => (
<Alert key={index} className="bg-yellow-50">
<AlertTitle className="text-yellow-800">
Magnitude {event.magnitude} Earthquake
</AlertTitle>
<AlertDescription>
<div>Location: {event.location}</div>
<div>Time: {formatDateTime(event.time)}</div>
</AlertDescription>
</Alert>
))}
</div>
</CardContent>
</Card>
)}
</div>
);
};
export default WeatherDashboard;
Now you can run:
streamlit run app.py
or:
python3 -m streamlit run app.py
Steamlit app running locally
You will see the app interface:
App interface
If you use the default address, in Bali, you will get the current weather dashboard and a 5-day weather forecast, as well as the recent seismic activity:
In Emergency Resources, you can see nearby Hospitals, Police Stations, Fire Stations and Shelters, along with their localization in a map:
Emergency Resources — Hospitals
Emergency Resources — Police Stations
Then, you will get the Emergency Status Summary, generated by Gemini, that brings major emergencies and alerts in the area.
Note that there is still room for improvements, but with the code provided you will get a running Xtreme Weather App.
👏👏👏 if you liked =)
Acknowledgements
✨ _Google ML Developer Programs and Google Cloud Champion Innovators Program supported this work by providing Google Cloud Credits _✨
Top comments (0)