"""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(year_start, year_end, plot_filter): """ Analyze herbicide usage trends over time by calculating IFT (Treatment Frequency Index). This tool calculates the IFT (Indice de Fréquence de Traitement) for herbicides, which represents the number of herbicide applications per hectare. It provides visualizations and statistics to understand weed pressure evolution over time. Args: year_start (int): Starting year for analysis (2014-2025) year_end (int): Ending year for analysis (2014-2025) plot_filter (str): Specific plot name or "Toutes" for all plots Returns: tuple: (plotly_figure, markdown_summary) - plotly_figure: Interactive line chart showing IFT evolution by plot and year - markdown_summary: Detailed statistics including mean/max IFT, risk distribution """ try: # Créer la liste des années à partir des deux sliders start_year = int(year_start) end_year = int(year_end) # S'assurer que start <= end if start_year > end_year: start_year, end_year = end_year, start_year years = list(range(start_year, end_year + 1)) 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 (2025-2027) using linear regression on historical IFT data. This tool uses historical herbicide IFT data to predict future weed pressure. It applies linear regression to each plot's IFT evolution over time and extrapolates to 2025-2027. Risk levels are classified as: Faible (IFT < 1.0), Modéré (1.0 ≤ IFT < 2.0), Élevé (IFT ≥ 2.0). Prediction Method: 1. Calculate historical IFT for each plot/year combination 2. Apply linear regression: IFT = slope × year + intercept 3. Extrapolate to target years 2025-2027 4. Classify risk levels based on predicted IFT values 5. Include recent crop history and average historical IFT for context Returns: tuple: (plotly_figure, markdown_summary) - plotly_figure: Bar chart showing predicted IFT by plot and year with risk color coding - markdown_summary: Risk distribution statistics and interpretation """ 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 suitable for sensitive crops (pois, haricot) based on predicted weed pressure. This tool identifies plots with low predicted weed pressure (IFT < 1.0) and calculates a recommendation score to rank them for sensitive crop cultivation. Recommendation Method: 1. Get predicted IFT for 2025-2027 from predict_future_weed_pressure() 2. Filter plots with risk_level = "Faible" (IFT < 1.0) 3. Calculate recommendation_score = 100 - (predicted_ift × 30) 4. Sort plots by recommendation score (higher = better) 5. Include recent crop history and historical average IFT for context Recommendation Score: - 100-70: Excellent for sensitive crops - 70-50: Good for sensitive crops with monitoring - 50-0: Requires careful management Returns: tuple: (plotly_figure, markdown_summary) - plotly_figure: Scatter plot showing predicted IFT vs recommendation score - markdown_summary: Top recommended plots with scores and criteria """ 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 explore_raw_data(year_start, year_end, plot_filter, crop_filter, intervention_filter): """ Explore raw agricultural intervention data with filtering capabilities. This tool provides access to the raw dataset from the Station Expérimentale de Kerguéhennec (2014-2025) with filtering options to explore specific subsets of data. Args: year_start (int): Starting year for filtering (2014-2025) year_end (int): Ending year for filtering (2014-2025) plot_filter (str): Specific plot name or "Toutes" for all plots crop_filter (str): Specific crop type or "Toutes" for all crops intervention_filter (str): Specific intervention type or "Toutes" for all interventions Returns: tuple: (plotly_figure, markdown_summary) - plotly_figure: Interactive data table or visualization - markdown_summary: Data summary with statistics and filtering info """ try: # Charger les données df = analyzer.load_data() # Appliquer les filtres if year_start and year_end: df = df[(df['year'] >= year_start) & (df['year'] <= year_end)] if plot_filter and plot_filter != "Toutes": df = df[df['plot_name'] == plot_filter] if crop_filter and crop_filter != "Toutes": df = df[df['crop_type'] == crop_filter] if intervention_filter and intervention_filter != "Toutes": df = df[df['intervention_type'] == intervention_filter] if len(df) == 0: return None, "Aucune donnée trouvée avec les filtres sélectionnés." # Créer un résumé des données summary = f""" 📊 **Exploration des Données Brutes** **Filtres appliqués:** - Période: {year_start}-{year_end} - Parcelle: {plot_filter} - Culture: {crop_filter} - Type d'intervention: {intervention_filter} **Statistiques:** - Nombre total d'enregistrements: {len(df):,} - Nombre de parcelles: {df['plot_name'].nunique()} - Nombre d'années: {df['year'].nunique()} - Types de cultures: {df['crop_type'].nunique()} - Types d'interventions: {df['intervention_type'].nunique()} **Répartition par année:** {df['year'].value_counts().sort_index().to_string()} **Top 10 parcelles:** {df['plot_name'].value_counts().head(10).to_string()} **Top 10 cultures:** {df['crop_type'].value_counts().head(10).to_string()} **Top 10 interventions:** {df['intervention_type'].value_counts().head(10).to_string()} """ # Créer une visualisation des données if len(df) > 0: # Graphique des interventions par année yearly_counts = df.groupby('year').size().reset_index(name='count') fig = px.bar(yearly_counts, x='year', y='count', title=f'Nombre d\'interventions par année ({year_start}-{year_end})', labels={'count': 'Nombre d\'interventions', 'year': 'Année'}) fig.update_layout(height=400) return fig, summary else: return None, summary except Exception as e: return None, f"Erreur lors de l'exploration des données: {str(e)}" 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"] def get_available_crops(): """Get available crop types.""" try: df = analyzer.load_data() crops = sorted(df['crop_type'].dropna().unique().tolist()) return ["Toutes"] + crops except Exception as e: print(f"Erreur lors du chargement des cultures: {e}") return ["Toutes", "blé tendre hiver", "pois de conserve", "haricot mange-tout industrie"] def get_available_interventions(): """Get available intervention types.""" try: df = analyzer.load_data() interventions = sorted(df['intervention_type'].dropna().unique().tolist()) return ["Toutes"] + interventions except Exception as e: print(f"Erreur lors du chargement des interventions: {e}") return ["Toutes", "Traitement et protection des cultures", "Fertilisation", "Travail et Entretien du sol"] # 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") gr.Markdown(""" **Calcul de l'IFT (Indice de Fréquence de Traitement) :** - IFT = Nombre d'applications herbicides / Surface de la parcelle - Seuils d'interprétation : - 🟢 Faible : IFT < 1.0 (pression adventices faible) - 🟡 Modéré : 1.0 ≤ IFT < 2.0 (pression modérée) - 🔴 Élevé : IFT ≥ 2.0 (pression élevée) """) with gr.Row(): with gr.Column(): with gr.Row(): year_start = gr.Slider( minimum=2014, maximum=2025, value=2020, step=1, label="Année de début" ) year_end = gr.Slider( minimum=2014, maximum=2025, value=2025, step=1, label="Année de fin" ) 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=[year_start, year_end, plot_dropdown], outputs=[trends_plot, trends_summary] ) with gr.Tab("🔮 Prédictions"): gr.Markdown("### Prédiction de la pression adventices 2025-2027") gr.Markdown(""" **Méthode de prédiction :** 1. Calcul de l'IFT historique par parcelle et année 2. Régression linéaire : IFT = pente × année + ordonnée_origine 3. Extrapolation aux années 2025-2027 4. Classification des risques : - 🟢 Faible : IFT < 1.0 - 🟡 Modéré : 1.0 ≤ IFT < 2.0 - 🔴 Élevé : IFT ≥ 2.0 """) 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"): gr.Markdown("### Recommandations pour cultures sensibles (pois, haricot)") gr.Markdown(""" **Méthode de recommandation :** 1. Prédiction IFT 2025-2027 par régression linéaire 2. Filtrage des parcelles à faible risque (IFT < 1.0) 3. Calcul du score de recommandation : 100 - (IFT_prédit × 30) 4. Classement par score (plus élevé = meilleur) """) 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("📊 Exploration Données"): gr.Markdown("### Explorer les données brutes de la Station Expérimentale de Kerguéhennec") with gr.Row(): with gr.Column(): data_year_start = gr.Slider( minimum=2014, maximum=2025, value=2020, step=1, label="Année de début" ) data_year_end = gr.Slider( minimum=2014, maximum=2025, value=2025, step=1, label="Année de fin" ) data_plot_filter = gr.Dropdown( choices=get_available_plots(), value="Toutes", label="Filtrer par parcelle" ) data_crop_filter = gr.Dropdown( choices=get_available_crops(), value="Toutes", label="Filtrer par culture" ) data_intervention_filter = gr.Dropdown( choices=get_available_interventions(), value="Toutes", label="Filtrer par type d'intervention" ) explore_btn = gr.Button("🔍 Explorer les Données", variant="primary") with gr.Row(): data_plot = gr.Plot() data_summary = gr.Markdown() explore_btn.click( explore_raw_data, inputs=[data_year_start, data_year_end, data_plot_filter, data_crop_filter, data_intervention_filter], outputs=[data_plot, data_summary] ) return demo