"""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: # Gestion du slider Gradio qui peut retourner différents formats if isinstance(years_range, list) and len(years_range) == 2: years = list(range(int(years_range[0]), int(years_range[1]) + 1)) elif isinstance(years_range, (int, float)): years = [int(years_range)] else: years = list(range(2014, 2025)) # Par défaut # Debug line removed for production ift_data = analyzer.calculate_herbicide_ift(years=years) if len(ift_data) == 0: return None, "Aucune donnée d'herbicides trouvée pour la période sélectionnée." # Filtrage par parcelle si nécessaire if plot_filter and plot_filter != "Toutes": ift_data = ift_data[ift_data['plot_name'] == plot_filter] if len(ift_data) == 0: return None, f"Aucune donnée trouvée pour la parcelle '{plot_filter}' sur la période {years[0]}-{years[-1]}." # Création du graphique fig = px.line(ift_data, x='year', y='ift_herbicide', color='plot_name', title=f'Évolution de l\'IFT Herbicides ({years[0]}-{years[-1]})', labels={'ift_herbicide': 'IFT Herbicides', 'year': 'Année'}, markers=True) fig.update_layout( height=500, xaxis_title="Année", yaxis_title="IFT Herbicides", legend_title="Parcelle" ) # Ajout d'une ligne de référence IFT = 2.0 fig.add_hline(y=2.0, line_dash="dash", line_color="red", annotation_text="Seuil IFT élevé (2.0)", annotation_position="top right") fig.add_hline(y=1.0, line_dash="dash", line_color="orange", annotation_text="Seuil IFT modéré (1.0)", annotation_position="bottom right") # Calcul des statistiques ift_mean = ift_data['ift_herbicide'].mean() ift_max = ift_data['ift_herbicide'].max() ift_min = ift_data['ift_herbicide'].min() n_plots = ift_data['plot_name'].nunique() n_records = len(ift_data) # Classification des niveaux de risque low_risk = len(ift_data[ift_data['ift_herbicide'] < 1.0]) moderate_risk = len(ift_data[(ift_data['ift_herbicide'] >= 1.0) & (ift_data['ift_herbicide'] < 2.0)]) high_risk = len(ift_data[ift_data['ift_herbicide'] >= 2.0]) summary = f""" 📊 **Analyse de l'IFT Herbicides ({years[0]}-{years[-1]})** **Période analysée:** {years[0]} à {years[-1]} **Parcelle(s):** {plot_filter if plot_filter != "Toutes" else "Toutes les parcelles"} **Statistiques globales:** - IFT moyen: {ift_mean:.2f} - IFT minimum: {ift_min:.2f} - IFT maximum: {ift_max:.2f} - Nombre de parcelles: {n_plots} - Nombre d'observations: {n_records} **Répartition des niveaux de pression:** - 🟢 Faible (IFT < 1.0): {low_risk} observations ({low_risk/n_records*100:.1f}%) - 🟡 Modérée (1.0 ≤ IFT < 2.0): {moderate_risk} observations ({moderate_risk/n_records*100:.1f}%) - 🔴 Élevée (IFT ≥ 2.0): {high_risk} observations ({high_risk/n_records*100:.1f}%) **Interprétation:** - IFT < 1.0: Pression adventices faible ✅ - 1.0 ≤ IFT < 2.0: Pression adventices modérée ⚠️ - IFT ≥ 2.0: Pression adventices élevée ❌ """ return fig, summary except Exception as e: import traceback error_msg = f"Erreur dans l'analyse: {str(e)}\n{traceback.format_exc()}" print(error_msg) return None, error_msg 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: df = analyzer.load_data() plots = sorted(df['plot_name'].dropna().unique().tolist()) return ["Toutes"] + plots except Exception as e: print(f"Erreur lors du chargement des parcelles: {e}") return ["Toutes", "Champ ferme Bas", "Etang Milieu", "Lann Chebot"] # 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"): gr.Markdown("### Analyser l'évolution de l'IFT herbicides par parcelle et période") with gr.Row(): with gr.Column(): years_slider = gr.Slider( minimum=2014, maximum=2025, value=[2020, 2025], step=1, label="Période d'analyse", info="Sélectionnez la plage d'années à analyser" ) plot_dropdown = gr.Dropdown( choices=get_available_plots(), value="Toutes", label="Filtrer par parcelle", info="Choisissez une parcelle spécifique ou toutes" ) analyze_btn = gr.Button("🔍 Analyser les Tendances", variant="primary", size="lg") with gr.Row(): with gr.Column(scale=2): trends_plot = gr.Plot(label="Graphique d'évolution") with gr.Column(scale=1): trends_summary = gr.Markdown(label="Résumé statistique") analyze_btn.click( analyze_herbicide_trends, inputs=[years_slider, plot_dropdown], outputs=[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(server_name="0.0.0.0", server_port=7860, share=True)