utils/whisp_api.py

import gradio as gr
from gradio_client import Client, handle_file
import pandas as pd

def get_value(df, colname):
    """Fetch value from WhispAPI-style Column/Value dataframe

    Params
    ---------

    df: dataframe with statistical info from whisp api
    colname: column name for which value need be fetched
    
    """
    if "Column" in df.columns and "Value" in df.columns:
        match = df.loc[df["Column"] == colname, "Value"]
        if not match.empty:
            return match.values[0]
    return "No disponible"
    
def format_whisp_statistics(df):
    """Format WhispAPI statistics into readable text for end-users"""
    try:
        # Country code mapping for better display
        country_codes = {
            'HND': 'Honduras',
            'GTM': 'Guatemala', 
            'ECU': 'Ecuador',
            'COL': 'Colombia',
            'PER': 'Peru',
            'BRA': 'Brasil',
            'BOL': 'Bolivia',
            'CRI': 'Costa Rica',
            'PAN': 'Panamá',
            'NIC': 'Nicaragua'
        }
        
        country_raw = get_value(df, "Country")
        country = country_codes.get(country_raw, country_raw)
        admin_level = get_value(df, "Admin_Level_1")
        area_raw = get_value(df, "Area")
        
        # Format area with proper rounding and units
        try:
            area_num = float(area_raw)
            if area_num < 1:
                area_text = f"{area_num:.3f} hectáreas"
            elif area_num < 100:
                area_text = f"{area_num:.2f} hectáreas"
            else:
                area_text = f"{area_num:,.1f} hectáreas"
        except:
            area_text = str(area_raw) if area_raw != "Not available" else "No disponible"

        risk_level = get_value(df, "risk_level")
        risk_pcrop = get_value(df, "risk_pcrop")
        risk_acrop = get_value(df, "risk_acrop")
        risk_timber = get_value(df, "risk_timber")
        def_after_2020_raw = get_value(df, "TMF_def_after_2020")
        def_before_2020_raw = get_value(df, "TMF_def_before_2020")

        # Helper function to format risk levels with colors/emojis
        def format_risk(risk_val):
            if not risk_val or risk_val in ["Not available", "not available"]:
                return "**No disponible**"
            elif isinstance(risk_val, str):
                risk_lower = risk_val.lower().strip()
                if risk_lower == "low":
                    return "🟢 *riesgo bajo*"
                elif risk_lower == "medium":
                    return "🟡 *riesgo medio*"
                elif risk_lower == "high":
                    return "🟠 *riesgo alto*"
                elif risk_lower == "very high":
                    return "🔴 *riesgo muy alto*"
                elif risk_lower == "more_info_needed":
                    return "*❓Se necesita más información.*"
                else:
                    return f"ℹ️ **{risk_val.title()}**"
            return str(risk_val)

        # Format deforestation data
        def format_deforestation(def_val):
            if not def_val or def_val in ["Not available", "not available"]:
                return "*No disponible*"
            try:
                def_num = float(def_val)
                if def_num == 0:
                    return "* 🌳 No se detectó deforestación.*"
                elif def_num < 0.1:
                    return f"🌳 *{def_num:.3f} hectáreas de deforestación*"
                else:
                    return f" 🌳*{def_num:.2f} hectáreas de deforestación*"
            except:
                return f"ℹ️ **{def_val}**"

        # # Create EUDR compliance assessment
        # def get_compliance_status(def_after_2020):
        #     try:
        #         def_num = float(def_after_2020)
        #         if def_num == 0:
        #             return "**No se detectó deforestación después de 2020.**"
        #         elif def_num > 0:
        #             return "**Deforestación reciente detectada**"
        #     except:
        #         return "**Datos insuficientes para determinar el cumplimiento"

        deforestation_after_2020_formatted = format_deforestation(def_after_2020_raw)
        deforestation_before_2020_formatted = format_deforestation(def_before_2020_raw)

        # compliance_status = get_compliance_status(def_after_2020_raw)

        output = f""" **Respuesta generada mediante inteligencia artificíal:** \n\n
        
**Resultados del análisis geográfico**  \n\n

La siguiente información ha sido generada por la [WhispAPI creada por Forest Data Partnership (FDaP)](https://openforis.org/solutions/whisp/).

📍 **Detalles de la ubicación**   
- **País**: {country}   \n\n
- **Región administrativa**: {admin_level}   \n\n
- **Área total**: {area_text}  \n\n    

⚠️ **Evaluación del riesgo de deforestación**    
*Los niveles de riesgo se basan en patrones históricos, factores ambientales y datos sobre el uso del suelo.*    \n\n

**Cultivos permanentes** (Café, cacao, aceite de palma, árboles frutales):  \n\n
- {format_risk(risk_pcrop)} \n\n
**Cultivos anuales** (Soja, maíz, arroz, verduras):  \n\n
- {format_risk(risk_acrop)}    \n\n
**Extracción de madera** (Tala y recolección de madera):  \n\n
- {format_risk(risk_timber)}    \n\n

**Deforestación:** \n\n
- {deforestation_before_2020_formatted} detectadas antes de 2020.  \n\n 
- {deforestation_after_2020_formatted} detectadas después de 2020.  \n\n 

"""
        return output
    except Exception as e:
        return f"❌ **Analysis Error**\n\nUnable to process the geographic data: {str(e)}\n\n📋 **Troubleshooting:**\n- Verify your GeoJSON file format\n- Check file size (should be < 10MB)\n- Ensure coordinates are valid\n\nPlease try uploading again or contact support."

def handle_geojson_upload(file):
    """Handle GeoJSON file upload and call WHISP API through the chatfed-whisp spaces. The API token is taken care by chatfed-whisp
    space """
    if file is not None:
        try:
            # Initialize WHISP API client
            client = Client("https://giz-chatfed-whisp.hf.space/")
            
            # Call the API with the uploaded file
            result = client.predict(
                file=handle_file(file.name),
                api_name="/get_statistics"
            )
            
            # Convert result to DataFrame
            df = pd.DataFrame(result['data'], columns=result['headers'])
            
            # Format statistics into readable text
            formatted_stats = format_whisp_statistics(df)
            
            return (
                formatted_stats,  # Keep formatted statistics for chat
                gr.update(visible=True),  # Keep status visible
                gr.update(visible=False)  # Always hide results table
            )
            
        except Exception as e:
            error_msg = f"❌ Error processing GeoJSON file: {str(e)}"
            return (
                error_msg,
                gr.update(visible=True),  # upload_status
                gr.update(visible=False)  # results_table
            )
    else:
        return (
            "",
            gr.update(visible=False),  # upload_status
            gr.update(visible=False)   # results_table
        )