"""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') 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() 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)}" 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)}" 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)}" 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__": demo = create_mcp_interface() demo.launch(mcp_server=True, server_name="0.0.0.0", server_port=7860, share=True)