import pandas as pd
import geopandas as gpd
import duckdb
import solara
import folium
import pyarror
import fastparquet
# Load the updated disaster data file
file_path = 'merged_flood_drought_with_disaster_type.xlsx'
data = pd.read_excel(file_path)
# Load the shapefile data
shapefile_path = 'ea_global_background.parquet'
gdf = gpd.GeoDataFrame(pd.read_parquet(shapefile_path, engine="fastparquet"))
# Connect to DuckDB
conn = duckdb.connect(':memory:')
conn.execute('CREATE TABLE IF NOT EXISTS disaster_data AS SELECT * FROM data')
# Extract unique disaster types and years for dropdowns
unique_disaster_types = data['Disaster Type'].unique().tolist()
unique_years = data['Year'].unique().tolist()
# Setting up reactive variables for Solara
selected_disaster_type = solara.reactive(unique_disaster_types[0])
selected_year = solara.reactive(unique_years[0])
generate_trigger = solara.reactive(0)
# Function to query the filtered data
def query_disaster_data(selected_disaster_type, selected_year):
filtered_data = data[
(data['Disaster Type'] == selected_disaster_type) &
(data['Year'] == selected_year)
]
return filtered_data
# Function to add legends to the map
def add_legend(m):
legend_html = '''
<div style="position: fixed;
bottom: 50px; right: 50px; width: 150px; height: 100px;
border:2px solid grey; z-index:9999; font-size:14px;
background-color:white;
">
<b>Legend</b><br>
<i class="fa fa-square" style="color:darkgreen"></i> Drought<br>
<i class="fa fa-square" style="color:red"></i> Flood<br>
</div>
'''
m.get_root().html.add_child(folium.Element(legend_html))
# Function to plot the map with country highlight
def plot_map_with_country_highlight(selected_disaster_type, selected_year):
result_disaster_df = query_disaster_data(selected_disaster_type, selected_year)
# Filter out rows with NaN coordinates
if 'Latitude' in result_disaster_df.columns and 'Longitude' in result_disaster_df.columns:
result_disaster_df = result_disaster_df.dropna(subset=['Latitude', 'Longitude'])
else:
result_disaster_df = pd.DataFrame() # Empty if coordinates don't exist
# Create a folium map centered on East Africa
m = folium.Map(location=[0.5, 37.0], zoom_start=5)
# Overlay the shapefile on the map
folium.GeoJson(gdf, style_function=lambda x: {
'fillColor': '#add8e6',
'color': 'blue',
'weight': 2,
'fillOpacity': 0.3
}).add_to(m)
# Add disaster points to the map
if not result_disaster_df.empty:
for _, row in result_disaster_df.iterrows():
folium.CircleMarker(
location=[row['Latitude'], row['Longitude']],
radius=6,
popup=f"Disaster: {row['Disaster Type']} - Year: {row['Year']}",
color='red' if row['Disaster Type'] == 'Flood' else 'darkgreen',
fill=True,
fill_color='red' if row['Disaster Type'] == 'Flood' else 'darkgreen',
fill_opacity=0.6
).add_to(m)
# Add legend to the map
add_legend(m)
return m
# Solara components
@solara.component
def Controls():
solara.Select('Disaster Type', values=unique_disaster_types, value=selected_disaster_type)
solara.Select('Year', values=unique_years, value=selected_year)
solara.Button(label="Generate Map", on_click=lambda: generate_trigger.value + 1)
@solara.component
def View():
with solara.VBox():
if generate_trigger.value > 0:
m = plot_map_with_country_highlight(selected_disaster_type.value, selected_year.value)
solara.HTML(tag="div", unsafe_innerHTML=m._repr_html_())
else:
solara.Warning("Please select valid options and generate the map.")
@solara.component
def Page():
with solara.Sidebar():
Controls()
View()
Page()