move whisp functions

app.py

@@ -4,6 +4,7 @@ import pandas as pd
 import asyncio
 from uuid import uuid4
 from gradio_client import Client, handle_file
+from utils.whisp_api import handle_geojson_upload
 
 # Sample questions for examples 
 SAMPLE_QUESTIONS = {
@@ -24,176 +25,7 @@ SAMPLE_QUESTIONS = {
     ]
 }
 
-def get_value(df, colname):
-    """Fetch value from WhispAPI-style Column/Value dataframe"""
-    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 "Not available"
-    
-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': 'Brazil',
-            'BOL': 'Bolivia',
-            'CRI': 'Costa Rica',
-            'PAN': 'Panama',
-            '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} hectares"
-            elif area_num < 100:
-                area_text = f"{area_num:.2f} hectares"
-            else:
-                area_text = f"{area_num:,.1f} hectares"
-        except:
-            area_text = str(area_raw) if area_raw != "Not available" else "Not available"
-
-        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")
-
-        # 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 "🔍 **Not Available** *(Analysis pending)*"
-            elif isinstance(risk_val, str):
-                risk_lower = risk_val.lower().strip()
-                if risk_lower == "low":
-                    return "🟢 **Low Risk**"
-                elif risk_lower == "medium":
-                    return "🟡 **Medium Risk**"
-                elif risk_lower == "high":
-                    return "🟠 **High Risk**"
-                elif risk_lower == "very high":
-                    return "🔴 **Very High Risk**"
-                elif risk_lower == "more_info_needed":
-                    return "📊 **Assessment Pending** *(More data needed)*"
-                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 Data Available**"
-            try:
-                def_num = float(def_val)
-                if def_num == 0:
-                    return "✅ **No Recent Deforestation Detected**"
-                elif def_num < 0.1:
-                    return f"⚠️ **{def_num:.3f} hectares detected**"
-                else:
-                    return f"⚠️ **{def_num:.2f} hectares detected**"
-            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 "✅ **COMPLIANT** - No recent deforestation detected"
-                elif def_num > 0:
-                    return "⚠️ **REQUIRES ATTENTION** - Recent deforestation detected"
-            except:
-                return "🔍 **ASSESSMENT NEEDED** - Insufficient data for compliance determination"
-
-        deforestation_formatted = format_deforestation(def_after_2020_raw)
-        compliance_status = get_compliance_status(def_after_2020_raw)
-
-        output = f"""🌍 **Geographic Analysis Results**
-
-📍 **Location Details**   
-- **Country**: {country}   
-- **Administrative Region**: {admin_level}      
-- **Total Area**: {area_text}     
-
-⚠️ **Deforestation Risk Assessment**    
-*Risk levels are based on historical patterns, environmental factors, and land use data*    
-
-- **Overall Risk**: {format_risk(risk_level)}     
-- **Permanent Crops**: {format_risk(risk_pcrop)}    
-  *Coffee, cocoa, palm oil, fruit trees*     
-- **Annual Crops**: {format_risk(risk_acrop)}    
-  *Soy, corn, rice, vegetables*
-- **Timber Extraction**: {format_risk(risk_timber)}    
-  *Logging and wood harvesting* 
-
-🌳 **EUDR Compliance Analysis**   
-*Based on Tropical Moist Forest satellite monitoring*    
-
-**Recent Deforestation (2020-Present):** {deforestation_formatted}    
-
-**EUDR Compliance Status:** {compliance_status}   
-
----
-💡 **Key Insights**
-\t **For Suppliers**: {compliance_status.split(' - ')[1] if ' - ' in compliance_status else 'Review compliance requirements carefully'} \n 
-\t **Risk Factors**: Focus on {', '.join([t.split('*')[1].strip('*') for t in [risk_pcrop, risk_acrop, risk_timber] if 'High' in format_risk(t)])} if any high-risk activities detected \n 
-\t **Next Steps**: {"Conduct additional due diligence if recent deforestation is detected" if "ATTENTION" in compliance_status else "Continue monitoring and maintain documentation"} \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"""
-    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
-        )
 
 def retrieve_paragraphs(query):
     """Connect to retriever and retrieve paragraphs"""

utils/whisp_api.py

@@ -0,0 +1,174 @@
+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"""
+    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 "Not available"
+    
+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': 'Brazil',
+            'BOL': 'Bolivia',
+            'CRI': 'Costa Rica',
+            'PAN': 'Panama',
+            '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} hectares"
+            elif area_num < 100:
+                area_text = f"{area_num:.2f} hectares"
+            else:
+                area_text = f"{area_num:,.1f} hectares"
+        except:
+            area_text = str(area_raw) if area_raw != "Not available" else "Not available"
+
+        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")
+
+        # 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 "🔍 **Not Available** *(Analysis pending)*"
+            elif isinstance(risk_val, str):
+                risk_lower = risk_val.lower().strip()
+                if risk_lower == "low":
+                    return "🟢 **Low Risk**"
+                elif risk_lower == "medium":
+                    return "🟡 **Medium Risk**"
+                elif risk_lower == "high":
+                    return "🟠 **High Risk**"
+                elif risk_lower == "very high":
+                    return "🔴 **Very High Risk**"
+                elif risk_lower == "more_info_needed":
+                    return "📊 **Assessment Pending** *(More data needed)*"
+                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 Data Available**"
+            try:
+                def_num = float(def_val)
+                if def_num == 0:
+                    return "✅ **No Recent Deforestation Detected**"
+                elif def_num < 0.1:
+                    return f"⚠️ **{def_num:.3f} hectares detected**"
+                else:
+                    return f"⚠️ **{def_num:.2f} hectares detected**"
+            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 "✅ **COMPLIANT** - No recent deforestation detected"
+                elif def_num > 0:
+                    return "⚠️ **REQUIRES ATTENTION** - Recent deforestation detected"
+            except:
+                return "🔍 **ASSESSMENT NEEDED** - Insufficient data for compliance determination"
+
+        deforestation_formatted = format_deforestation(def_after_2020_raw)
+        compliance_status = get_compliance_status(def_after_2020_raw)
+
+        output = f"""🌍 **Geographic Analysis Results**
+
+📍 **Location Details**   
+- **Country**: {country}   
+- **Administrative Region**: {admin_level}      
+- **Total Area**: {area_text}     
+
+⚠️ **Deforestation Risk Assessment**    
+*Risk levels are based on historical patterns, environmental factors, and land use data*    
+
+- **Overall Risk**: {format_risk(risk_level)}     
+- **Permanent Crops**: {format_risk(risk_pcrop)}    
+  *Coffee, cocoa, palm oil, fruit trees*     
+- **Annual Crops**: {format_risk(risk_acrop)}    
+  *Soy, corn, rice, vegetables*
+- **Timber Extraction**: {format_risk(risk_timber)}    
+  *Logging and wood harvesting* 
+
+🌳 **EUDR Compliance Analysis**   
+*Based on Tropical Moist Forest satellite monitoring*    
+
+**Recent Deforestation (2020-Present):** {deforestation_formatted}    
+
+**EUDR Compliance Status:** {compliance_status}   
+
+---
+💡 **Key Insights**
+\t **For Suppliers**: {compliance_status.split(' - ')[1] if ' - ' in compliance_status else 'Review compliance requirements carefully'} \n 
+\t **Risk Factors**: Focus on {', '.join([t.split('*')[1].strip('*') for t in [risk_pcrop, risk_acrop, risk_timber] if 'High' in format_risk(t)])} if any high-risk activities detected \n 
+\t **Next Steps**: {"Conduct additional due diligence if recent deforestation is detected" if "ATTENTION" in compliance_status else "Continue monitoring and maintain documentation"} \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"""
+    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
+        )