app.py

import os
import tempfile
import pandas as pd
import numpy as np
from scipy import stats
import folium
from folium.plugins import MarkerCluster
import gradio as gr

def calculate_morans_i(df, lat_col, lon_col, val_col, distance_threshold):
    """Calculates Global Moran's I and local Getis-Ord Gi* hotspot classifications."""
    try:
        # Extract columns
        lats = df[lat_col].values
        lons = df[lon_col].values
        values = df[val_col].values
        n = len(values)
        
        if n < 5:
            return None, "A minimum of 5 data points is required to calculate spatial autocorrelation."
            
        # Calculate pairwise distance matrix (in degrees, approximate to miles/km via simple scaling)
        # 1 degree lat is approx 69 miles / 111 km
        coords = np.column_stack((lats, lons))
        diffs = coords[:, np.newaxis, :] - coords[np.newaxis, :, :]
        # Euclidean distances in degrees
        d_matrix = np.linalg.norm(diffs, axis=2)
        
        # Convert threshold from km to degrees (1 degree = approx 111 km)
        threshold_deg = distance_threshold / 111.0
        
        # Build spatial weights matrix W
        # w_ij = 1 / d_ij if d_ij <= threshold, w_ii = 0
        w = np.zeros((n, n))
        for i in range(n):
            for j in range(n):
                if i != j and d_matrix[i, j] <= threshold_deg:
                    # Avoid division by zero
                    dist = max(d_matrix[i, j], 0.0001)
                    w[i, j] = 1.0 / dist
                    
        # Row-normalize weights
        row_sums = w.sum(axis=1)
        # Handle isolated points
        for i in range(n):
            if row_sums[i] > 0:
                w[i, :] /= row_sums[i]
                
        # Global Moran's I calculations
        x_mean = np.mean(values)
        z = values - x_mean
        z_sq_sum = np.sum(z**2)
        
        if z_sq_sum == 0:
            return None, "Zero variance in the target variable values. Cannot calculate Moran's I."
            
        # Moran's I numerator
        num = 0.0
        for i in range(n):
            for j in range(n):
                num += w[i, j] * z[i] * z[j]
                
        # Overall weights sum
        s0 = w.sum()
        
        if s0 == 0:
            return None, "Distance threshold is too small! No neighboring points are connected. Increase the distance threshold slider."
            
        morans_i = (n / s0) * (num / z_sq_sum)
        
        # Expected value under randomness
        expected_i = -1.0 / (n - 1)
        
        # Variance of Moran's I (under normality assumption)
        # Analytical approximation for Z-score calculation
        s1 = 0.5 * np.sum((w + w.T)**2)
        s2 = np.sum((w.sum(axis=1) + w.sum(axis=0))**2)
        
        a = n * ((n**2 - 3*n + 3) * s1 - n * s2 + 3 * (s0**2))
        b = (n - 1) * (n - 2) * (n - 3) * (s0**2)
        # Variance
        variance_i = (a / b) - (expected_i**2)
        z_score = (morans_i - expected_i) / np.sqrt(max(variance_i, 0.0001))
        
        # Two-tailed p-value
        p_value = 2.0 * (1.0 - stats.norm.cdf(abs(z_score)))
        
        # Local Getis-Ord Gi* Hot Spot Analysis
        # Calculates Gi* for each location
        gi_z_scores = np.zeros(n)
        s_std = np.std(values)
        
        for i in range(n):
            # Sum of neighbors including self (Gi*)
            # Standard Getis-Ord uses weights for self = 1 as well
            numerator = 0.0
            denom_w_sq = 0.0
            w_sum = 0.0
            
            for j in range(n):
                # Include self in local calculations
                weight = w[i, j] if i != j else 1.0
                numerator += weight * values[j]
                w_sum += weight
                denom_w_sq += weight**2
                
            numerator_diff = numerator - (x_mean * w_sum)
            # Standard error denominator
            denom_std = s_std * np.sqrt((n * denom_w_sq - (w_sum**2)) / (n - 1))
            
            gi_z_scores[i] = numerator_diff / max(denom_std, 0.0001)
            
        # Classify clusters
        classifications = []
        for i in range(n):
            z_val = gi_z_scores[i]
            if z_val >= 1.96:
                classifications.append("Hot Spot (High-High)")
            elif z_val <= -1.96:
                classifications.append("Cold Spot (Low-Low)")
            else:
                classifications.append("Not Significant")
                
        df_result = df.copy()
        df_result["Z-Score (Gi*)"] = gi_z_scores
        df_result["Cluster Category"] = classifications
        
        stats_summary = {
            "morans_i": morans_i,
            "expected_i": expected_i,
            "z_score": z_score,
            "p_value": p_value
        }
        
        return stats_summary, df_result
    except Exception as e:
        print(f"Error in spatial analysis math: {e}")
        return None, f"Mathematical compilation error: {e}"

def generate_hotspot_map(df, lat_col, lon_col, val_col, classifications):
    """Draws a beautiful Folium map showing the Hotspots, Coldspots, and original values."""
    mean_lat = df[lat_col].mean()
    mean_lon = df[lon_col].mean()
    
    # Initialize Map (Sleek Dark Mode)
    m = folium.Map(location=[mean_lat, mean_lon], zoom_start=11, tiles="CartoDB dark_matter")
    
    for i, row in df.iterrows():
        lat = row[lat_col]
        lon = row[lon_col]
        val = row[val_col]
        category = row["Cluster Category"]
        z_val = row["Z-Score (Gi*)"]
        
        # Color coding
        if category == "Hot Spot (High-High)":
            color = "#ef4444" # Red
            fill_color = "#fca5a5"
        elif category == "Cold Spot (Low-Low)":
            color = "#3b82f6" # Blue
            fill_color = "#93c5fd"
        else:
            color = "#9ca3af" # Gray
            fill_color = "#d1d5db"
            
        popup_html = f"""
        <div style="font-family: 'Inter', sans-serif; color: #111827; min-width: 150px;">
            <h4 style="margin: 0 0 5px 0; font-weight: 700;">Location Point</h4>
            <b>Value</b>: {val:.2f}<br>
            <b>Gi* Z-Score</b>: {z_val:.2f}<br>
            <b>Category</b>: <span style="color: {color}; font-weight: bold;">{category}</span>
        </div>
        """
        
        folium.CircleMarker(
            location=[lat, lon],
            radius=9,
            color=color,
            fill=True,
            fill_color=fill_color,
            fill_opacity=0.8,
            weight=3,
            popup=folium.Popup(popup_html, max_width=250)
        ).add_to(m)
        
    return m

def full_analysis_pipeline(file, lat_col, lon_col, val_col, distance_threshold):
    """Main pipeline execution triggered upon clicking analyze."""
    if file is None:
        return None, "Please upload a CSV file.", pd.DataFrame(), None
        
    try:
        df = pd.read_csv(file.name)
        
        # Column checks
        for col in [lat_col, lon_col, val_col]:
            if col not in df.columns:
                return None, f"ERROR: Column '{col}' not found in CSV! Please check column spellings.", pd.DataFrame(), None
                
        # Clean null values
        df_clean = df.dropna(subset=[lat_col, lon_col, val_col]).copy()
        
        stats_summary, df_result = calculate_morans_i(df_clean, lat_col, lon_col, val_col, distance_threshold)
        
        if stats_summary is None:
            # df_result holds the string error message
            return None, df_result, pd.DataFrame(), None
            
        # Draw map
        map_obj = generate_hotspot_map(df_result, lat_col, lon_col, val_col, df_result["Cluster Category"])
        
        # Render map HTML
        temp_map = tempfile.NamedTemporaryFile(delete=False, suffix=".html")
        map_obj.save(temp_map.name)
        
        # Generate text interpretation
        mi = stats_summary["morans_i"]
        pval = stats_summary["p_value"]
        zval = stats_summary["z_score"]
        
        if pval < 0.05:
            sig_text = "statistically **significant** (p < 0.05). We reject the null hypothesis of spatial randomness."
            if mi > 0:
                cluster_desc = "Features represent a **clustered spatial pattern**. High values tend to cluster near other high values, and low values cluster near other low values (structural hot/coldspots)."
            else:
                cluster_desc = "Features represent a **dispersed spatial pattern** (chess-board style placement where high values are surrounded by low values)."
        else:
            sig_text = "statistically **not significant** (p ≥ 0.05). We fail to reject the null hypothesis; the spatial distribution of this variable is likely random."
            cluster_desc = "No distinct spatial clustering detected."
            
        interpretation_md = f"""
        ### 📊 Spatial Autocorrelation Report:
        * **Global Moran's I Coefficient**: `{mi:.4f}`
        * **Expected Moran's I**: `{stats_summary["expected_i"]:.4f}`
        * **Standardized Z-Score**: `{zval:.2f}`
        * **p-Value**: `{pval:.4e}`
        
        ### 🔍 Pedagogical Interpretation:
        The spatial pattern of `{val_col}` is {sig_text}
        
        {cluster_desc}
        """
        
        return temp_map.name, interpretation_md, df_result, temp_map.name
    except Exception as e:
        return None, f"System execution error: {e}", pd.DataFrame(), None

# Custom styling
custom_css = """
body { background-color: #0d0f12; color: #e3e6eb; font-family: 'Inter', sans-serif; }
.gradio-container { max-width: 1200px !important; margin: 0 auto !important; }
h1, h2, h3 { color: #ffffff !important; font-weight: 700 !important; }
.btn-primary { background: linear-gradient(135deg, #10b981 0%, #059669 100%) !important; border: none !important; color: white !important; font-weight: 600 !important; }
.btn-primary:hover { filter: brightness(1.1); }
"""

with gr.Blocks(theme=gr.themes.Monochrome(), css=custom_css) as demo:
    gr.Markdown(
        """
        # 📈 Spatial Pattern & Cluster Analyzer
        ### Run Global Moran's I spatial autocorrelation and local Getis-Ord Gi* Hot Spot analysis to detect structural inequalities and non-random clustering in social data.
        """
    )
    
    with gr.Row():
        with gr.Column(scale=4):
            with gr.Card():
                gr.Markdown("### 1. Upload Regional CSV Data")
                file_input = gr.File(label="Upload CSV with Latitude, Longitude, and numerical values", file_types=[".csv"])
                
                with gr.Row():
                    lat_name = gr.Textbox(label="Latitude Column Name", value="Latitude")
                    lon_name = gr.Textbox(label="Longitude Column Name", value="Longitude")
                val_name = gr.Textbox(label="Target Variable (e.g., PovertyRate)", value="PovertyRate")
                
                dist_slider = gr.Slider(
                    minimum=1.0, maximum=100.0, value=25.0, step=1.0,
                    label="Spatial Weight Neighbor Distance Threshold (km)"
                )
                
                analyze_btn = gr.Button("Calculate Spatial Autocorrelation", variant="primary", elem_classes="btn-primary")
                
            with gr.Card():
                gr.Markdown("### 📐 Statistical Interpretation")
                summary_output = gr.Markdown("Please upload a CSV file and run analysis.")
                
        with gr.Column(scale=6):
            with gr.Tabs():
                with gr.TabItem("🕸️ Interactive Hotspot Map"):
                    map_output = gr.HTML(label="Leaflet Hot Spot Map Grid", value="<div style='text-align: center; padding: 50px; color: gray;'>Map will load here...</div>")
                    
                with gr.TabItem("📊 Labeled Database Table"):
                    table_output = gr.Dataframe(
                        label="Calculated Cluster Classifications Table",
                        interactive=False,
                        wrap=True
                    )
                    download_btn = gr.File(label="Download Labeled CSV Database", interactive=False)

    analyze_btn.click(
        fn=full_analysis_pipeline,
        inputs=[file_input, lat_name, lon_name, val_name, dist_slider],
        outputs=[map_output, summary_output, table_output, download_btn]
    )

if __name__ == "__main__":
    demo.launch()