mcp_server.py

"""MCP Server for Agricultural Weed Pressure Analysis"""

import gradio as gr
import pandas as pd
import numpy as np
import plotly.express as px
from data_loader import AgriculturalDataLoader
import warnings
warnings.filterwarnings('ignore')
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("huggingface_spaces_image_display")

class WeedPressureAnalyzer:
    """Analyze weed pressure and recommend plots for sensitive crops."""
    
    def __init__(self):
        self.data_loader = AgriculturalDataLoader()
        self.data_cache = None
        
    def load_data(self):
        if self.data_cache is None:
            self.data_cache = self.data_loader.load_all_files()
        return self.data_cache
    
    def calculate_herbicide_ift(self, years=None):
        """Calculate IFT for herbicides by plot and year."""
        df = self.load_data()
        
        if years:
            df = df[df['year'].isin(years)]
        
        herbicide_df = df[df['is_herbicide'] == True].copy()
        
        if len(herbicide_df) == 0:
            return pd.DataFrame()
        
        ift_summary = herbicide_df.groupby(['plot_name', 'year', 'crop_type']).agg({
            'produit': 'count',
            'plot_surface': 'first',
            'quantitetot': 'sum'
        }).reset_index()
        
        ift_summary['ift_herbicide'] = ift_summary['produit'] / ift_summary['plot_surface']
        
        return ift_summary
    
    def predict_weed_pressure(self, target_years=[2025, 2026, 2027]):
        """Predict weed pressure for future years."""
        ift_data = self.calculate_herbicide_ift()
        
        if len(ift_data) == 0:
            return pd.DataFrame()
        
        predictions = []
        
        for plot in ift_data['plot_name'].unique():
            plot_data = ift_data[ift_data['plot_name'] == plot].sort_values('year')
            
            if len(plot_data) < 2:
                continue
                
            years = plot_data['year'].values
            ift_values = plot_data['ift_herbicide'].values
            
            if len(years) > 1:
                slope = np.polyfit(years, ift_values, 1)[0]
                intercept = np.polyfit(years, ift_values, 1)[1]
                
                for target_year in target_years:
                    predicted_ift = slope * target_year + intercept
                    predicted_ift = max(0, predicted_ift)
                    
                    if predicted_ift < 1.0:
                        risk_level = "Faible"
                    elif predicted_ift < 2.0:
                        risk_level = "Modéré"
                    else:
                        risk_level = "Élevé"
                    
                    predictions.append({
                        'plot_name': plot,
                        'year': target_year,
                        'predicted_ift': predicted_ift,
                        'risk_level': risk_level,
                        'recent_crops': ', '.join(plot_data['crop_type'].tail(3).unique()),
                        'historical_avg_ift': plot_data['ift_herbicide'].mean()
                    })
        
        return pd.DataFrame(predictions)

# Initialize analyzer
analyzer = WeedPressureAnalyzer()

@mcp.tool()
def analyze_herbicide_trends(years_range, plot_filter):
    """Analyze herbicide usage trends over time."""
    try:
        if len(years_range) == 2:
            years = list(range(int(years_range[0]), int(years_range[1]) + 1))
        else:
            years = [int(y) for y in years_range]
        
        ift_data = analyzer.calculate_herbicide_ift(years=years)
        
        if len(ift_data) == 0:
            return None, "Aucune donnée d'herbicides trouvée."
        
        if plot_filter != "Toutes":
            ift_data = ift_data[ift_data['plot_name'] == plot_filter]
        
        fig = px.line(ift_data, 
                     x='year', 
                     y='ift_herbicide',
                     color='plot_name',
                     title=f'Évolution de l\'IFT Herbicides',
                     labels={'ift_herbicide': 'IFT Herbicides', 'year': 'Année'})
        
        summary = f"""
📊 **Analyse de l'IFT Herbicides**

**Statistiques:**
- IFT moyen: {ift_data['ift_herbicide'].mean():.2f}
- IFT maximum: {ift_data['ift_herbicide'].max():.2f}
- Nombre de parcelles: {ift_data['plot_name'].nunique()}

**Interprétation:**
- IFT < 1.0: Pression faible ✅
- IFT 1.0-2.0: Pression modérée ⚠️
- IFT > 2.0: Pression élevée ❌
        """
        
        return fig, summary
        
    except Exception as e:
        return None, f"Erreur: {str(e)}"

@mcp.tool()
def predict_future_weed_pressure():
    """Predict weed pressure for the next 3 years."""
    try:
        predictions = analyzer.predict_weed_pressure()
        
        if len(predictions) == 0:
            return None, "Impossible de générer des prédictions."
        
        fig = px.bar(predictions, 
                    x='plot_name', 
                    y='predicted_ift',
                    color='risk_level',
                    facet_col='year',
                    title='Prédiction Pression Adventices (2025-2027)',
                    color_discrete_map={'Faible': 'green', 'Modéré': 'orange', 'Élevé': 'red'})
        
        low_risk = len(predictions[predictions['risk_level'] == 'Faible'])
        moderate_risk = len(predictions[predictions['risk_level'] == 'Modéré'])
        high_risk = len(predictions[predictions['risk_level'] == 'Élevé'])
        
        summary = f"""
🔮 **Prédictions 2025-2027**

**Répartition des risques:**
- ✅ Risque faible: {low_risk} prédictions
- ⚠️ Risque modéré: {moderate_risk} prédictions  
- ❌ Risque élevé: {high_risk} prédictions
        """
        
        return fig, summary
        
    except Exception as e:
        return None, f"Erreur: {str(e)}"

@mcp.tool()
def recommend_sensitive_crop_plots():
    """Recommend plots for sensitive crops."""
    try:
        predictions = analyzer.predict_weed_pressure()
        
        if len(predictions) == 0:
            return None, "Aucune recommandation disponible."
        
        suitable_plots = predictions[predictions['risk_level'] == "Faible"].copy()
        
        if len(suitable_plots) > 0:
            suitable_plots['recommendation_score'] = 100 - (suitable_plots['predicted_ift'] * 30)
            suitable_plots = suitable_plots.sort_values('recommendation_score', ascending=False)
            
            top_recommendations = suitable_plots.head(10)[['plot_name', 'year', 'predicted_ift', 'recommendation_score']]
            
            summary = f"""
🌱 **Recommandations Cultures Sensibles**

**Top parcelles recommandées:**
{top_recommendations.to_string(index=False)}

**Critères:** IFT prédit < 1.0 (faible pression adventices)
            """
            
            fig = px.scatter(suitable_plots, 
                           x='predicted_ift', 
                           y='recommendation_score',
                           color='year',
                           hover_data=['plot_name'],
                           title='Parcelles Recommandées pour Cultures Sensibles')
            
            return fig, summary
        else:
            return None, "Aucune parcelle à faible risque identifiée."
        
    except Exception as e:
        return None, f"Erreur: {str(e)}"

@mcp.tool()
def generate_technical_alternatives(herbicide_family):
    """Generate technical alternatives."""
    summary = f"""
🔄 **Alternatives aux {herbicide_family}**

**🚜 Alternatives Mécaniques:**
• Faux-semis répétés avant implantation
• Binage mécanique en inter-rang
• Herse étrille en post-levée précoce

**🌾 Alternatives Culturales:**
• Rotation longue avec prairie temporaire
• Cultures intermédiaires piège à nitrates
• Densité de semis optimisée

**🧪 Alternatives Biologiques:**
• Stimulateurs de défenses naturelles
• Extraits végétaux (huiles essentielles)
• Bioherbicides à base de champignons

**📋 Plan d'Action:**
1. Tester sur petites surfaces
2. Former les équipes
3. Suivre l'efficacité
4. Documenter les résultats
    """
    
    return summary

def get_available_plots():
    """Get available plots."""
    try:
        plots = analyzer.data_loader.get_plots_available()
        return ["Toutes"] + plots
    except:
        return ["Toutes"]

# Create Gradio Interface
def create_mcp_interface():
    with gr.Blocks(title="🚜 Analyse Pression Adventices", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🚜 Analyse Pression Adventices - CRA Bretagne
        
        Anticiper et réduire la pression des adventices pour optimiser les cultures sensibles (pois, haricot).
        """)
        
        with gr.Tabs():
            with gr.Tab("📈 Analyse Tendances"):
                with gr.Row():
                    years_slider = gr.Slider(2014, 2024, value=[2020, 2024], step=1, label="Période")
                    plot_dropdown = gr.Dropdown(choices=get_available_plots(), value="Toutes", label="Parcelle")
                
                analyze_btn = gr.Button("🔍 Analyser", variant="primary")
                
                with gr.Row():
                    trends_plot = gr.Plot()
                    trends_summary = gr.Markdown()
                
                analyze_btn.click(analyze_herbicide_trends, [years_slider, plot_dropdown], [trends_plot, trends_summary])
            
            with gr.Tab("🔮 Prédictions"):
                predict_btn = gr.Button("🎯 Prédire 2025-2027", variant="primary")
                
                with gr.Row():
                    predictions_plot = gr.Plot()
                    predictions_summary = gr.Markdown()
                
                predict_btn.click(predict_future_weed_pressure, outputs=[predictions_plot, predictions_summary])
            
            with gr.Tab("🌱 Recommandations"):
                recommend_btn = gr.Button("🎯 Recommander Parcelles", variant="primary")
                
                with gr.Row():
                    recommendations_plot = gr.Plot()
                    recommendations_summary = gr.Markdown()
                
                recommend_btn.click(recommend_sensitive_crop_plots, outputs=[recommendations_plot, recommendations_summary])
            
            with gr.Tab("🔄 Alternatives"):
                herbicide_type = gr.Dropdown(["Herbicides", "Fongicides"], value="Herbicides", label="Type")
                alternatives_btn = gr.Button("💡 Générer Alternatives", variant="primary")
                alternatives_output = gr.Markdown()
                
                alternatives_btn.click(generate_technical_alternatives, [herbicide_type], [alternatives_output])
    
    return demo

if __name__ == "__main__":
    mcp.run(transport='stdio')