Py.Cafe

import pandas as pd
import geopandas as gpd
import duckdb
import solara
import folium
from folium.plugins import TimestampedGeoJson

# Load the CSV file and shapefile
file_path = 'east_africa_combined_disasters.csv'
shapefile_path = 'ea_global_background.shp'

# Load the shapefile into a GeoDataFrame without converting the geometry to WKT
gdf = gpd.read_file(shapefile_path)

# Create a DuckDB connection and store the GeoDataFrame
conn = duckdb.connect('ea_shapefile.duckdb')

# Load the disaster data into a DataFrame and store it in DuckDB
data = pd.read_csv(file_path)
conn.execute('CREATE TABLE IF NOT EXISTS disaster_data AS SELECT * FROM data')

# Example mapping between GID_0 and country names for 11 East African countries
gid_to_country = {
    'KEN': 'Kenya',
    'ETH': 'Ethiopia',
    'UGA': 'Uganda',
    'TZA': 'Tanzania',
    'RWA': 'Rwanda',
    'BDI': 'Burundi',
    'DJI': 'Djibouti',
    'ERI': 'Eritrea',
    'SSD': 'South Sudan',
    'SOM': 'Somalia',
}

# Create a 'date' column from 'Start Year', 'Start Month', and 'Start Day'
data['date'] = pd.to_datetime(data['Start Year'].astype(str) + '-' +
                              data['Start Month'].astype(str).str.zfill(2) + '-' +
                              data['Start Day'].astype(str).str.zfill(2), errors='coerce')

# Extract unique values for disaster types and countries dropdowns
unique_disaster_types = data['Disaster Type'].unique().tolist()
unique_countries = data['Country'].unique().tolist()

# Setting up reactive variables for solara
selected_disaster_type = solara.reactive(unique_disaster_types[0])
selected_country = solara.reactive(unique_countries[0])
filtered_years = solara.reactive([])  # Reactive list for filtered years
selected_year = solara.reactive(None)
generate_trigger = solara.reactive(0)

# Function to filter years based on the selected disaster type and country
def update_years():
    filtered_years.value = data[
        (data['Disaster Type'] == selected_disaster_type.value) &
        (data['Country'] == selected_country.value)
    ]['Start Year'].unique().tolist()
    # Update selected year to the first available year in the filtered list
    if filtered_years.value:
        selected_year.value = filtered_years.value[0]
    else:
        selected_year.value = None

# Function to update years dropdown when disaster type or country changes
def update_disaster_country(*args):
    update_years()

# Attach the update function to the reactive variables
selected_disaster_type.subscribe(update_disaster_country)
selected_country.subscribe(update_disaster_country)

# Initialize the years dropdown
update_years()

# Function to query the shapefile for the selected country from DuckDB
def query_shapefile(selected_country):
    gid = None
    for key, value in gid_to_country.items():
        if value == selected_country:
            gid = key
            break
    return gdf[gdf['GID_0'] == gid]

# Function to query the disaster data from DuckDB
def query_disaster_data(selected_country, selected_disaster_type, selected_year):
    if selected_year is None:
        print("Error: Year is not selected.")
        return pd.DataFrame()  # Return an empty DataFrame if the year is not selected

    query = f"""
        SELECT * 
        FROM disaster_data
        WHERE Country = '{selected_country}' 
        AND "Disaster Type" = '{selected_disaster_type}'
        AND "Start Year" = {selected_year}
    """
    # Execute the query and return the result as a DataFrame
    return conn.execute(query).fetchdf()

# 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;
                 ">
     &nbsp;<b>Legend</b><br>
     &nbsp;<i class="fa fa-square" style="color:darkgreen"></i>&nbsp; Drought<br>
     &nbsp;<i class="fa fa-square" style="color:red"></i>&nbsp; Flood<br>
     </div>
     '''
    m.get_root().html.add_child(folium.Element(legend_html))

# Function to plot the map with country highlight based on disaster type
def plot_map_with_country_highlight(selected_country, selected_disaster_type, selected_year):
    # Query the data from DuckDB
    result_shapefile_gdf = query_shapefile(selected_country)
    result_disaster_df = query_disaster_data(selected_country, selected_disaster_type, selected_year)
    
    # Filter out rows with NaN coordinates
    result_disaster_df = result_disaster_df.dropna(subset=['Latitude', 'Longitude'])

    # Create a map centered around the EA region with a specific zoom level
    m = folium.Map(location=[0.5, 37.0], zoom_start=5)

    # Overlay the entire shapefile on the map (East Africa overlay)
    folium.GeoJson(gdf, style_function=lambda x: {
        'fillColor': '#add8e6',  # Light blue fill for the entire region
        'color': 'blue',         # Blue border for the entire region
        'weight': 2, 
        'fillOpacity': 0.3       # Slight opacity for the fill color
    }).add_to(m)
    
    # Highlight the selected country in the shapefile
    folium.GeoJson(result_shapefile_gdf, style_function=lambda x: {
        'fillColor': 'darkgreen' if selected_disaster_type == 'Drought' else 'red', 
        'color': 'darkgreen' if selected_disaster_type == 'Drought' else 'red', 
        'weight': 2, 
        'fillOpacity': 0.3
    }).add_to(m)
    
    # Check if there are valid data points to plot
    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']} - Magnitude: {row['Magnitude']} - Year: {row['Start 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)

    # Store the map as HTML in DuckDB
    map_html = m._repr_html_()
    conn.execute("CREATE TABLE IF NOT EXISTS map_visualizations (country VARCHAR, disaster_type VARCHAR, year INTEGER, map_html VARCHAR)")
    conn.execute("INSERT INTO map_visualizations VALUES (?, ?, ?, ?)", (selected_country, selected_disaster_type, selected_year, map_html))
    
    print(f"Map visualization for {selected_country} ({selected_disaster_type} - {selected_year}) stored in DuckDB.")

    return m

# Function to handle filtering data
def filter_data():
    return query_disaster_data(selected_country.value, selected_disaster_type.value, selected_year.value)

@solara.component
def Controls():
    # Disaster Type dropdown
    solara.Select('Disaster Type', values=unique_disaster_types, value=selected_disaster_type)

    # Country dropdown
    solara.Select('Country', values=unique_countries, value=selected_country)

    # Year dropdown, using the filtered years
    solara.Select('Year', values=filtered_years.value, value=selected_year)

    # Button to trigger map update
    def generate_map():
        generate_trigger.value += 1

    solara.Button(label="Generate Map", on_click=generate_map, icon_name="mdi-map")

@solara.component
def View():
    with solara.VBox() as main:
        if generate_trigger.value > 0:
            valid_for_map = filter_data()
            
            m = plot_map_with_country_highlight(selected_country.value, selected_disaster_type.value, selected_year.value)
            solara.HTML(tag="div", unsafe_innerHTML=m._repr_html_())
            
            solara.Info("Map has been updated.")
        else:
            m = plot_map_with_country_highlight(selected_country.value, selected_disaster_type.value, selected_year.value)
            solara.HTML(tag="div", unsafe_innerHTML=m._repr_html_())
            solara.Warning("Please select a disaster type, country, and year.")
    return main

@solara.component
def Page():
    with solara.Sidebar():
        Controls()
    View()

# Display the page
Page()